结合 RxTextView Observable 和 Retrofit Observable [英] Combine RxTextView Observable and Retrofit Observable

问题描述

作为开始使用 RxAndroid 的示例，我正在尝试实现一个搜索框，当用户插入某些内容时，该搜索框会触发休息调用.

As an example to getting started with RxAndroid I'm trying to implement a searchbox which triggers a rest call when the users inserts something.

到目前为止，我有两个工作部分.第一次观察EditTextView ...

So far I have two working parts. The first observing the EditTextView ...

RxTextView.textChangeEvents(searchEditText)
    .debounce(400, TimeUnit.MILLISECONDS)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Observer<TextViewTextChangeEvent>() {
            @Override
            public void onCompleted() {
                Timber.d("onCompleted");
            }

            @Override
            public void onError(Throwable e) {
                Timber.e(e, "onError");
                }

            @Override
            public void onNext(TextViewTextChangeEvent e) {
                Timber.d("onNext" + e.text().toString());
            }
        });

... 第二部分使用改造服务调用 REST API:

... and the second part calling the REST API by using a Retrofit Service:

APIManager.getService().searchRestaurants("test")
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer<List<Restaurant>>() {
            @Override
            public void onCompleted() {
                Timber.d("onCompleted");
            }

            @Override
            public void onError(Throwable e) {
                Timber.e(e, "onError");
            }

            @Override
            public void onNext(List<Restaurant> restaurants) {
                Timber.d("onNext");
                for (Restaurant restaurant : restaurants) {
                    Timber.d(restaurant.getId() + ": " + restaurant.getName());
                }
            }
        });

我的问题是将这两部分结合起来.我尝试使用 flatMap 运算符如下:

My Problem is combining the two parts. I tried by using the flatMap Operator as following:

RxTextView.textChangeEvents(searchEditText)
        .debounce(400, TimeUnit.MILLISECONDS)
        .observeOn(AndroidSchedulers.mainThread())
        .flatMap(new Func1<TextViewTextChangeEvent, Observable<List<Restaurant>>>() {
            @Override
            public Observable<List<Restaurant>> call(TextViewTextChangeEvent txtChangeEvt) {
                return APIManager.getService().searchRestaurants(txtChangeEvt.text().toString());
            }
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer<List<Restaurant>>() {
            @Override
            public void onCompleted() {
                Timber.d("onCompleted");
            }

            @Override
            public void onError(Throwable e) {
                Timber.e(e, "onError");
            }

            @Override
            public void onNext(List<Restaurant> restaurants) {
                Timber.d("onNext");
                for (Restaurant restaurant : restaurants) {
                    Timber.d(restaurant.getId() + ": " + restaurant.getName());
                }
            }
        });

当我这样做时，我得到以下异常:

When I do this I get following exception:

java.lang.IllegalStateException: Must be called from the main thread. Was: Thread[RxCachedThreadScheduler-1,5,main]
                                                                              at com.jakewharton.rxbinding.internal.Preconditions.checkUiThread(Preconditions.java:28)
                                                                              at com.jakewharton.rxbinding.widget.TextViewTextChangeEventOnSubscribe.call(TextViewTextChangeEventOnSubscribe.java:21)
                                                                              at com.jakewharton.rxbinding.widget.TextViewTextChangeEventOnSubscribe.call(TextViewTextChangeEventOnSubscribe.java:12)

所以我试图通过调用 subscribeOn(AndroidSchedulers.mainThread() 来解决这个问题，但在这种情况下，当然，我得到了一个 NetworkOnMainThread 异常.

So I tried to fix that by calling subscribeOn(AndroidSchedulers.mainThread() but in this case, of course, I get an NetworkOnMainThread Exception.

那我该怎么做呢?组合应该在不同线程上执行的不同 Observable 的正确方法是什么?

So how Do I do this? What is a proper way to combine different Observables which should execute on different Threads?

推荐答案

只需删除第一个 .observeOn(AndroidSchedulers.mainThread()).看看这个示例

Just remove the first .observeOn(AndroidSchedulers.mainThread()). Take a look at this example

Observable.just(1) // 1 will be emited in the IO thread pool
    .subscribeOn(Schedulers.io())
    .flatMap(...) // will be in the IO thread pool
    .observeOn(Schedulers.computation())
    .flatMap(...) // will be executed in the computation thread pool
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(); // will be executed in the Android main thread (if you're running your code on Android)

这篇关于结合 RxTextView Observable 和 Retrofit Observable的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！