Rx Java mergeDelayError funktioniert nicht wie erwartet

Question

Ich benutze RxJava in und Android-Anwendung mit RxAndroid. Ich verwende mergeDelayError, um zwei Retro-Fit-Netzwerkaufrufe zu einer Observablen zu kombinieren, die ausgegebene Elemente verarbeiten wird, wenn eines ausgegeben wird, und den Fehler, wenn es eins hat. Dies funktioniert nicht und es wird nur die Aktion onError ausgelöst, wenn ein Fehler auftritt. Nun, um das zu testen, habe ich ein sehr einfaches Beispiel gewählt und immer noch wird die successAction nie aufgerufen, wenn ich einen onError-Aufruf habe. Siehe Beispiel unten.

Die Erfolgsaktion wird nur aufgerufen, wenn ich zwei Erfolgsobservablen verwende. Fehle ich etwas mit wie mergeDelayError funktionieren soll?

EDIT:

ich gefunden habe, dass, wenn ich die observeOn und subscribeOn alles entfernen wie erwartet funktioniert. Ich muss Threads angeben und dachte, dass das der Sinn von Rx ist. Irgendeine Idee, warum das Spezifizieren dieser Schedulers das Verhalten brechen würde?

Answer 1

Dies scheint immer noch wie ein Fehler im MergeDelayError-Operator, aber ich konnte es durch Duplizieren der ObserverOn und Subscribe für jede Observable arbeiten.

Observable.mergeDelayError(
      Observable.error(new RuntimeException()) 
       .observeOn(AndroidSchedulers.mainThread()) 
       .subscribeOn(Schedulers.io()), 
      Observable.just("Hello") 
       .observeOn(AndroidSchedulers.mainThread()) 
       .subscribeOn(Schedulers.io()) 
     ) 
     .finallyDo(completeAction) 
     .subscribe(successAction, errorAction); 

Answer 2

Ich glaube, Sie nicht für das Terminal Ereignis warten und der Haupt-Thread beendet, bevor die Ereignisse zu Ihrem Beobachter geliefert werden. Der folgende Test geht für mich mit RxJava 1.0.14:

@Test 
public void errorDelayed() { 
    TestSubscriber<Object> ts = TestSubscriber.create(); 
    Observable.mergeDelayError(
      Observable.error(new RuntimeException()), 
      Observable.just("Hello") 
     ) 
     .subscribeOn(Schedulers.io()).subscribe(ts); 

    ts.awaitTerminalEvent(); 

    ts.assertError(RuntimeException.class); 
    ts.assertValue("Hello"); 
} 

Answer 3

Verwenden .observeOn (AndroidSchedulers.mainThread(), true) statt .observeOn (AndroidSchedulers.mainThread()

public final Observable<T> observeOn(Scheduler scheduler, boolean delayError) { 
     return observeOn(scheduler, delayError, RxRingBuffer.SIZE); 
    } 

Oben ist die Unterschrift von observeOn Funktion. Code Werke folgen.

Observable.mergeDelayError(
       Observable.error(new RuntimeException()), 
       Observable.just("Hello") 
     ) 
       .observeOn(AndroidSchedulers.mainThread(), true) 
       .subscribeOn(Schedulers.io()) 
       .subscribe(new Subscriber<String>() { 
        @Override 
        public void onCompleted() { 

        } 

        @Override 
        public void onError(Throwable e) { 

        } 

        @Override 
        public void onNext(String s) { 

        } 
       }); 

diesen Trick von C Got oncatDelayError thread: https://github.com/ReactiveX/RxJava/issues/3908#issuecomment-217999009