RxJava onCompleted e onTerminate sul thread principale

Question

Sto utilizzando RxJava con Retrofit 2.0 su Android per le richieste di rete.

Quando sto creando il osservabile, aggiungo il seguente ad esso:

observable = observable 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribeOn(Schedulers.io()) 
    .unsubscribeOn(Schedulers.io()) 

Se poi aggiungo una:

observable = observable.doOnTerminate(new Action0() { 
       @Override 
       public void call() { 
        Log.d("OBS", "[" + Thread.currentThread().getName() + "] onTerminate"); 
       } 
      }); 

o simili per doOnError e doOnCompleted i callback sono eseguiti su il thread IO, mentre doOnNext viene eseguito sul thread principale.

Tuttavia, quello che voglio veramente è che tutti i callback vadano al thread principale, ma l'esecuzione dovrebbe rimanere sul thread IO.

C'è una soluzione elegante a questo senza dover avvolgere manualmente le mie implementazioni con un blocco per pubblicare qualcosa sul thread principale?

Answer 1

Si dovrebbe posizionare i callback prima di ogni observeOn così rimarranno sul suo thread precedente:

Observable.range(1, 10) 
.subscribeOn(Schedulers.io()) 
.doOnTerminate(() -> System.out.println(Thread.currentThread())) 
.map(v -> v + 1) 
.observeOn(AndroidSchedulers.mainThread()) 
.map(v -> Thread.currentThread() + "/" + v) 
.doOnNext(v -> Log.d("OBS", v)) 
.subscribe();