AndroidObservable从不在主线程上观察

Question

我正在Android上试验RxJava。我正在尝试一个简单的例子，在那里我试图卸载一个更长的操作在后台线程，然后期望回到主线程，做一个ui操作。

我从EditText onTextChangeListener(通过一个漂亮的ButterKnife注入)获得搜索文本，然后将其发送到一个长时间运行的操作_searchForContacts(searchText)，该操作返回一个联系人对象列表。然后，我通过在适配器上设置信息并通知数据集更改来更新我的视图。

然而，当我试图做任何ui更新时，我遇到了一个异常，因为我从来没有在主线程上。

下面是我的代码:

public class MyTestFragment
    extends Fragment
    implements Observer<List<Contact>> {
    // ...
    private Subscription _searchGuestsSubscription = Subscriptions.empty();
    // ...
    @Override
    public void onDestroyView() {
        super.onDestroyView();
        _searchGuestsSubscription.unsubscribe();
    }
    @Override
    public void onActivityCreated(Bundle savedInstanceState) {
        super.onActivityCreated(savedInstanceState);
        _adapter = new ContactImportAdapter();
        _adapter.setContacts(_searchForContacts(ALL_CONTACTS));
        _listView.setAdapter(_adapter);
    }
    @Override
    public void onNext(List<Contact> contactSearchResults) {
        Timber.d("I'm on the main thread -> " + String.valueOf(Looper.myLooper() == Looper.getMainLooper()));
        _adapter.setContacts(contactSearchResults);
    }
    @Override
    public void onCompleted() {
        _updateView();
    }
    @Override
    public void onError(Throwable e) {
        Timber.e(e, "Oops something went wrong.");
    }
    // ButterKnife ~ onTextChangeListener
    @OnTextChanged(R.id.search_edit_text)
    void onSearchContact(CharSequence searchChars) {
        _searchGuestsSubscription = AndroidObservable.bindFragment(this,
                                                                   _searchGuestsObservable(searchChars.toString()))
                                                     // The below line doesn't seem to have any effect?
                                                     .observeOn(AndroidSchedulers.mainThread())
                                                     .subscribeOn(Schedulers.io())
                                                     .subscribe(this);
    }
    private Observable<List<Contact>> _searchGuestsObservable(final String searchText) {
        return Observable.create(new Observable.OnSubscribe<List<Contact>>() {

            @Override
            public void call(Subscriber<? super List<Contact>> searchResultObserver) {
                // do the search
                List<Contact> contactSearchResults = _searchForContacts(searchText);
                onNext(contactSearchResults);
                onCompleted();
            }
        });
    }
    private void _updateView() {
        _adapter.notifyDataSetChanged();
    }
}

它说onNext总是在后台线程上被调用。我得到了通常的IllegalStateException: The current thread must have a looper!异常。日志语句也表明我不在onNext的主线程上。

给了什么?

Answer 1

您的问题似乎来自这样一个事实，即您直接在Fragment中调用onNext，而不是在订阅者上调用onNext。

像这样修改代码

private Observable<List<Contact>> _searchGuestsObservable(final String searchText) {
    return Observable.create(new Observable.OnSubscribe<List<Contact>>() {
        @Override
        public void call(Subscriber<? super List<Contact>> searchResultObserver) {
            // do the search
            List<Contact> contactSearchResults = _searchForContacts(searchText);
            searchResultObserver.onNext(contactSearchResults);
            searchResultObserver.onCompleted();
        }
    });
}

当你的Observable.OnSubscribe<List<Contact>>()被执行时，你实际上是在后台线程上，因为subscribeOn(Schedulers.io())，你正在调用你的Fragment的onNext自己，这就是为什么你得到那个异常。