rxjava catch

¿Cómo funcionan RxJava doOnError y onErrorReturn?

Question

Hice estas pruebas unitarias, y el resultado no es lo que esperaba en absoluto:

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

Entonces, tanto la salida "subscribe.onError" - ni doOnError ni onErrorReturn parecen ser llamados.

doOnError se documenta como:

Modifica el origen observable para que invoque una acción si llama a error.

No estoy seguro de cómo interpretar eso, pero esperaba que "doOnError" se imprimiera o "doOnError" seguido de "subscribe.onError".

onErrorReturn se documenta como:

Indica a un Observable que emita un elemento (devuelto por una función específica) en lugar de invocar a Onrror si encuentra un error.

Por lo tanto, esperaba "got: Yeah I got this" como resultado de la última prueba.

¿Lo que da?

Answer 1

Tanto doOnError como onErrorReturn devuelven un nuevo Observable con el comportamiento modificado. Estoy de acuerdo en que la documentación de ellos puede ser un poco engañosa. Modifique sus pruebas como esta para obtener el comportamiento esperado:

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