RxJava2 .subscribeOn .observeOn confusion. Running on Main thread

The order in which you put observeOn() and sunbscribeOn() , and other operators is very important.

• subscribeOn() operator tells the source Observable which thread to emit and transform items on.

• Be careful where you put the observeOn() operator because it changesthe thread performing the work! In most cases you probably want to delay switching to the observing thread until the very end of your Rx chain.

  Observable.just("long", "longer", "longest")    .subscribeOn(Schedulers.computation())    .observeOn(AndroidSchedulers.mainThread())    .map(String::length)    .filter(length -> length == 6)    .subscribe(length -> System.out.println("item length " + length));

here the map, filter and consuming is performed in the main thread

• observeOn() before map()There is no reason to have observeOn() operator applied above the map() operator. In fact, this code will result in NetworkOnMainThreadException! We do not want to be reading from HTTP response on the main thread — it should be done before we switch back to the main thread.

• you can also use multiple observeOn() to switch threads like this example.

   Observable.just("long", "longer", "longest").doOnNext(s -> System.out.println("first doOnNext: processing item on thread " + Thread.currentThread().getName())).observeOn(Schedulers.computation()).map(String::toString).doOnNext(s -> System.out.println("second doOnNext: processing item on thread " + Thread.currentThread().getName())).observeOn(Schedulers.io()).map(String::toString).subscribeOn(Schedulers.newThread()).map(String::length).subscribe(length -> System.out.println("received item length " + length + " on thread " + Thread.currentThread().getName()));

OUTPUT :

first doOnNext: processing item on thread RxNewThreadScheduler-1first doOnNext: processing item on thread RxNewThreadScheduler-1first doOnNext: processing item on thread RxNewThreadScheduler-1second doOnNext: processing item on thread RxComputationThreadPool-1second doOnNext: processing item on thread RxComputationThreadPool-1second doOnNext: processing item on thread RxComputationThreadPool-1received item length 4 on thread RxCachedThreadScheduler-1received item length 6 on thread RxCachedThreadScheduler-1received item length 7 on thread RxCachedThreadScheduler-1

Noteaccording to this answer subscribeOn() does't apply to the downstream operators, Therefore it does not guarantee that your operation is going to be on a different Thread.

subscribeOn effects go upstream and closer to the source of events.

As for your problem I have made a test and here are the results

   private void testRxJava2Async() { io.reactivex.Observable.fromCallable(new Callable<String>() {        @Override        public String call() throws Exception {            Log.d(TAG,"callable (expensive assync method) was called on --> "+Thread.currentThread().getName());            return null;        }    })            .subscribeOn(Schedulers.io())            .observeOn(AndroidSchedulers.mainThread())            .subscribeWith(new Observer<String>() {                @Override                public void onSubscribe(Disposable d) {                }                @Override                public void onNext(String s) {                    Log.d(TAG,"onNext() was called on --> "+Thread.currentThread().getName());                }                @Override                public void onError(Throwable e) {                }                @Override                public void onComplete() {                }            });}

Test Result :

callable (expensive assync method) was called on --> RxCachedThreadScheduler-1onNext() was called on--> main

Believe me i understand your confusion, let me explain step by step. Remember this for every subscribe there should be observe. We all know all subscribe should be executed on a worker thread(IO) and observer on Main thread(UI). In the first example the subscribe fired, then you said, for this subscription all updates will be delivered on the main thread, regardless of any more subscriptions during any transformations. However in the second example , this is what you said , observe only when the map transformations on the subscriptions is transformed then run on Main thread. Think of it, as Pre and Post increment in programming.

Hope this makes sense. RX framework is great. but actually requires live practice to get it right. Has you have seen first hand.

To ensure your changes are pushed and performed on the main thread what you need to do is to add a intermediary step to observe the changes.

Integer[] list = {6,3,2,1}; Observable.just(list).subscribeOn(Schedulers.newThread()).observeOn(Schedulers.newThread()).map(value = > value  * 2).observeOn(Schedulers.mainThread()).subscribeWith(...)

The idea is this subscribeOn only accepts to start the processing , such as a NetworkRequest but does not guarantee that values will be pushed to that same thread. observeOn says hey i can receive those values you initially subscribed for. So to make sure the values are transformed on the main thread, you can observe changes on another plain thread , perform a operation(Map| Flatmap etc), then observe those changes this will guarantee only those values are placed on the main thread. Basically what your saying is this, hey perform all the heavy processing in those threads, but hey whenever the computations are done, pass me those values to me in the main thread to continue my work, only then can i accept, therefore no work will ever be done on the main thread.