onComplete() is not getting called on the specified Scheduler

[sadegh]

when I use Observable.take(time, TimeUnit.SECONDS), if the result Observable stops due to timeout, onComplete() callback gets called on computation thread, not the one that I've specified with observeOn().

Here is a sample code:

Observable.just("one", "two")
        .observeOn(Schedulers.io())
        .take(0, TimeUnit.SECONDS)
        .subscribe(new Observer<String>() {
            @Override public void onSubscribe(Disposable d) {
                System.out.println("onSubscribe(), thread: " + Thread.currentThread().getName());
            }

            @Override public void onNext(String s) {
                System.out.println("onNext(), thread: " + Thread.currentThread().getName());
            }

            @Override public void onError(Throwable e) {
                System.out.println("onError(), thread: " + Thread.currentThread().getName());
            }

            @Override public void onComplete() {
                System.out.println("onComplete, thread: " + Thread.currentThread().getName());
            }
        });

And as the result you'll get onComplete, thread: RxComputationThreadPool-1.

If it's required for all callback methods to run on the specified thread per Reactive specifications, this would be a bug, and it happens on both RxJava 1.x and 2.x

[akarnokd]

observeOn lasts until the next async operator, in this case take. You can specify the Scheduler where take will time out. Alternatively, you can swap take and observeOn so events will end up on the desired thread.

[sadegh]

Yeah, I looked into the code and figured what's causing that. However I believe this is kinda inconsistent behavior, (and probably wrong), since all onNext() calls are running on the specified scheduler, and all of the sudden onComplete() runs on another thread.

At least we should have this behavior documented. It's not known that take() might change thread for some callbacks! (onError is also possible to run on computation)

[sadegh]

I also looked into the source code and figured why this is happening.
Here is my quick (and probably naive!) fix that I did to just test my idea about the source of this issue.
I added an if to Observable.take()

public final Observable<T> take(long time, TimeUnit unit) {
  if (this instanceof ObservableObserveOn) 
    return takeUntil(timer(time, unit)).observeOn(((ObservableObserveOn) this).scheduler);
  else 
    return takeUntil(timer(time, unit));
}

[akarnokd]

The operator behaves and expected. If the upstream completes before the time runs out, those events will be delivered on the upstream's thread. If there is a timeout, which is scheduled by a different Scheduler, the terminal event will be delivered on that other thread. If you want to ensure events are delivered on the desired thread, apply observeOn just before that.

If you want, you can create a PR wich add a clause to the relevant operators' Javadoc (in both Observable and Flowable which explicitly states the completion triggered by the timeout may come from a different thread (depending on the Scheduler used) than the items came from.

[sadegh]

Alright, I'll add it to doc.

[akarnokd]

Closing via #5718 and #5719.