How does RxJava doOnError and onErrorReturn work?

Question

I made these unit tests, and the outcome is not what I expected at all:

// This one outputs "subscribe.onError"  @Test public void observable_doOnError_subscribingToError() throws InterruptedException {     Observable<String> obs = getErrorProducingObservable();     obs.doOnError(throwable -> System.out.println("doOnError"));     obs.subscribeOn(Schedulers.immediate()).observeOn(Schedulers.immediate()).subscribe(         s -> {},         error -> System.out.println("subscribe.onError")     );     Thread.sleep(300); }  // This one outputs "subscribe.onError"  @Test public void observable_onErrorReturn() throws InterruptedException {     Observable<String> obs = getErrorProducingObservable();     obs.onErrorReturn(throwable -> "Yeah I got this");     obs.subscribeOn(Schedulers.immediate()).observeOn(Schedulers.immediate()).subscribe(         s -> System.out.println("got: " + s),         error -> System.out.println("subscribe.onError")     );     Thread.sleep(300); }  private Observable<String> getErrorProducingObservable()  {     return Observable.create(subscriber -> {         subscriber.onError(new RuntimeException("Somebody set up us the bomb"));     }); } 

So both output "subscribe.onError" - neither doOnError nor onErrorReturn seems to be called.

doOnErroris documented as:

Modifies the source Observable so that it invokes an action if it calls onError.

I'm not sure how to intepret that, but I expected either "doOnError" to be output or "doOnError" followed by "subscribe.onError".

onErrorReturn is documented as :

Instructs an Observable to emit an item (returned by a specified function) rather than invoking onError if it encounters an error.

Hence I was expecting "got: Yeah I got this" as output from the latter test.

What gives?

Answer 1

Both doOnError and onErrorReturn returns a new Observable with the changed behaviour. I agree that the documentation of them may be a little misleading. Modify your tests like this to get the expected behaviour:

// This one outputs "subscribe.onError"  @Test public void observable_doOnError_subscribingToError() throws InterruptedException {     Observable<String> obs =          getErrorProducingObservable()             .doOnError(throwable -> System.out.println("doOnError"));      obs.subscribeOn(Schedulers.immediate()).observeOn(Schedulers.immediate()).subscribe(         s -> {},         error -> System.out.println("subscribe.onError")     );     Thread.sleep(300); }  // This one outputs "subscribe.onError"  @Test public void observable_onErrorReturn() throws InterruptedException {     Observable<String> obs =          getErrorProducingObservable()             .onErrorReturn(throwable -> "Yeah I got this");      obs.subscribeOn(Schedulers.immediate()).observeOn(Schedulers.immediate()).subscribe(         s -> System.out.println("got: " + s),         error -> System.out.println("subscribe.onError")     );     Thread.sleep(300); }  private Observable<String> getErrorProducingObservable()  {     return Observable.create(subscriber -> {         subscriber.onError(new RuntimeException("Somebody set up us the bomb"));     }); }