Schedulers for doOnSubscribe(), finallyDo(), etc.

[Daniel Lew]

Hello,

Currently it seems somewhat like undefined behavior on which thread methods like doOnSubscribe() and finallyDo() act. As far as I can tell, doOnSubscribe() uses the Scheduler defined in subscribeOn(), whereas finallyDo() uses observeOn(). But what if the behavior desired here is to have both act on a separate Scheduler?

Simple example of this causing issues for me: increment a value when an Observable is subscribed to, decrement afterwards. If value is ever > 0, then we show a progress bar. I need these to run on particular threads on Android or it'll blow up (manipulating UI from non-main thread).

I'm wondering if I'm just not using these methods properly - it does seem a bit like introducing side effects - but then, what is the point of these methods otherwise?

-Dan

[Ben Christensen]

Those operators don’t use any schedulers. They behave just like `map`, they synchronously execute on whatever thread calls it. So `doOnSubscribe` is on the thread doing the subscribing and `doOnNext`/`doOnCompleted`/`doOnError`/`finallyDo` execute on whatever thread emits those signals. 

Yes, the point of those methods is side-effects (such as logging, debugging, or can be used as you are using it).

-- 
Ben Christensen
+1.310.782.5511  @benjchristensen

[Daniel Lew]

I guess then that it just won't work in this case, then? It's fairly common pattern to want to subscribe on something like Schedulers.io() but execute on AndroidSchedulers.mainThread(), since you can't modify the UI from anything but the main thread. That works fine for finallyDo() (and the rest) but not for doOnSubscribe().

-Dan

[Ben Christensen]

doOnSubscribe by definition is happening “onSubscribe” and thus on that thread. If it is already side-effecting (which it is) then you could emit to another Observable that uses observeOn to put itself on the UI thread. I don’t like Subjects at all, but this may be an example where it applies since you are purposefully doing a side-effect that is not composed inside the stream (you’re reaching outside anyways, so using a Subject wouldn’t be much different).

Or, you could use just the AndroidSchedulers.mainThread() directly and schedule work on to it all you’re doing is scheduling an update to the UI and not actually using the Observable behaviors.

Would be interesting to ask Erik Meijer’s opinion on this … I’ll see what his thoughts are.

-- 
Ben Christensen
+1.310.782.5511  @benjchristensen

[Dmitry Suzdalev]

Yep, I was going to write about the same post.

In Android it is quite useful to do usual chain-like building up of observable items stream and at some point do some side-effecting stuff like updating some UI title or label, which needs to happen on a main thread.

I also thought that current way of doing this might be by using and AndroidSchedulers.mainThread() directly. 

Your example with a subject is good too, thanks.

[Ben Christensen]

I got confirmation from Erik Meijer that this is one of those times that a Subject is indeed a good option. 

[moluae...@gmail.com]

I have found out that the position of doOnSubcribe in the chain matters. It hurts my mind to think of a reason. This doesn't work

matchRepository.
        commenceMatch(new NewMatch(matchId)).

        doOnSubscribe(disposable -> homeView.showProgress(true, true)).
        observeOn(SchedulerProvider.getInstance().ui()).

        subscribeOn(SchedulerProvider.getInstance().computation()).

        doFinally(() -> homeView.showProgress(false, false)).
        subscribe(() -> {
            homeView.showWaitingScene();
        }, this::handleError);

I get a toast saying that you cannot manipulate views from another thread. However

matchRepository.
        commenceMatch(new NewMatch(matchId)).
        observeOn(SchedulerProvider.getInstance().ui()).
        subscribeOn(SchedulerProvider.getInstance().computation()).
        doOnSubscribe(disposable -> homeView.showProgress(true, true)).
        doFinally(() -> homeView.showProgress(false, false)).
        subscribe(() -> {
            homeView.showWaitingScene();
        }, this::handleError);

This works. It is as if the execution of the chain starts even before the subcribe is called.