android – 结合RxTextView Observable和Retrofit Observable

作为开始使用RxAndroid的一个例子,我试图实现一个搜索框,当用户插入一些内容时触发一个休息调用.

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

RxTextView.textChangeEvents(searchEditText)
    .debounce(400,TimeUnit.MILLISECONDS)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Observer

…以及使用Retrofit Service调用REST API的第二部分：

apimanager.getService().searchRestaurants("test")
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer

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

RxTextView.textChangeEvents(searchEditText)
        .debounce(400,TimeUnit.MILLISECONDS)
        .observeOn(AndroidSchedulers.mainThread())
        .flatMap(new Func1apimanager.getService().searchRestaurants(txtChangeEvt.text().toString());
            }
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer

当我这样做时,我得到以下异常：

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()来解决这个问题,但在这种情况下,当然,我得到了一个N​​etworkOnMainThread异常.

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

最佳答案

只需删除第一个.observeOn(AndroidSchedulers.mainThread()).看看这个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)