Combine RxTextView Observable y Retrofit Observable

Como ejemplo para empezar con RxAndroid, estoy intentando implementar un searchbox que activa una llamada de reposo cuando los usuarios insertan algo.

Hasta ahora tengo dos partes de trabajo. La primera observando la EditTextView …

• Utilizando RxJava para la validación de inicio de sesión de correo electrónico, un observable está emitiendo dos veces
• No vuelva a ejecutar la llamada de Retrofit si todavía está en curso con RxJava 2
• RxJava y Retrofit2: NetworkOnMainThreadException
• La comprensión incorrecta del búfer en RxJava
• RxBindings para Spinner?

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()); } }); 

… y la segunda parte que llama a la API REST mediante el uso de un Servicio de Retrofit:

 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()); } } }); 

Mi problema es combinar las dos partes. Intenté usar el operador del flatMap como sigue:

 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()); } } }); 

Cuando hago esto obtengo la siguiente excepción:

 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) 

Así que traté de solucionarlo llamando a subscribeOn(AndroidSchedulers.mainThread() pero en este caso, por supuesto, obtengo una excepción de NetworkOnMainThread.

Entonces, ¿cómo hago esto? ¿Cuál es una manera apropiada de combinar diferentes Observables que deben ejecutarse en diferentes Hilos?

 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)