Come funziona RxJava doOnError e onErrorReturn?

Question

Ho fatto questi test di unità, e il risultato non è quello che mi aspettavo affatto:

// 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")); 
    }); 
} 

Così sia uscita "subscribe.onError" - né doOnError né onErrorReturn sembra di essere chiamato.

doOnError è documentato come:

modifica la sorgente osservabile in modo che richiama un'azione se chiama onError.

io non sono sicuro di come intepret che, ma mi aspettavo uno "doOnError" da emettere o "doOnError" seguito da "subscribe.onError".

onErrorReturn è documentato come:

incarica un'osservabile per emettere una voce (restituito da una funzione specificata) anziché invocare onError se si verifica un errore.

Quindi mi aspettavo "ottenuto: Sì, ho ottenuto questo" come risultato di quest'ultimo test.

Cosa dà?

Answer 1

Entrambi doOnError e onErrorReturn restituiscono un nuovo Observable con il comportamento modificato. Sono d'accordo che la documentazione di loro possa essere un po 'fuorviante. Modifica i tuoi test in questo modo per ottenere il comportamento previsto:

// 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")); 
    }); 
}