Applying schedulers twice in an observed chain (using layout)

Question

I have several network calls in my application. I like to run a network request in the input / output stream using the link operator with this transformer:

public static <T> Transformer<T, T> runOnIoThread() { return tObservable -> tObservable.subscribeOn( Schedulers.io() ) .observeOn( AndroidSchedulers.mainThread() ); } 

This seems to work well as long as I only have unified network calls. However, if I bind them, as in the following example, I get an Android NetworkInMainThreadException.

 public Observable<String> networkCall1() { return <NETWORK_CALL_1()> .compose( runOnIoThread() ); } public Observable<String> networkCall2( String input ) { return <NETWORK_CALL_2(input)> .compose( runOnIoThread() ); } public Observable<String> chainedCalls() { return networkCall1() .flatMap( result1 -> networkCall2( result1 ) ); } 

Earlier, my idea was that compose is applied to the entire observed chain before the call, and subsequent calls to compose will “overwrite” the behavior of the previous one. In fact, it looks like the observeOn call of the first compose ( observeOn main thread) dominates the second compose call ( subscribeOn IO thread). One obvious solution would be to have two versions of networkCall1 — one that schedulers use and the other not. However, this will make my code rather verbose.

Do you know the best solutions? Can you explain the behavior of the use of planners twice (with compilation) in the observed chain?

Edit: I am using a modification with RxJava for my network calls.

Answer 1

You can only use subscribeOn() once for each stream. If you use it a second time, it will not do anything. Thus, when you combine your two methods together, you run:

observeOn(AndroidSchedulers.mainThread())

which switches the operation to the main thread. After that, it stays there because the next subscribeOn() actually ignored.

I would suggest that you are actually too complex with your layout method. Just add

subscribeOn(Schedulers.io())

For your network calls, then use

observeOn(AndroidSchedulers.mainThread())

Before you want to process the results in the main thread. You will get something like:

 public Observable<String> networkCall1() { return <NETWORK_CALL_1()> .subscribeOn(Schedulers.io); } public Observable<String> networkCall2( String input ) { return <NETWORK_CALL_2(input)> .subscribeOn(Schedulers.io); } public Observable<String> chainedCalls() { return networkCall1() .flatMap( result1 -> networkCall2( result1 ) ) .observeOn(AndroidSchedulers.mainThread()); } 

EDIT

If you really want to have a call to observeOn() for specific group call methods. You need to add extra observeOn() to your chainedCalls() method. You can have as many observeOn() calls as you want for each thread. It will be something like:

 public Observable<String> networkCall1() { return <NETWORK_CALL_1()> .subscribeOn(Schedulers.io) .observeOn(AndroidSchedulers.mainThread()); } public Observable<String> networkCall2( String input ) { return <NETWORK_CALL_2(input)> .subscribeOn(Schedulers.io) .observeOn(AndroidSchedulers.mainThread()); } public Observable<String> chainedCalls() { return networkCall1() .observeOn(Schedulers.io) .flatMap( result1 -> networkCall2( result1 ) ) .observeOn(AndroidSchedulers.mainThread()); }