RxJava onErrorResumeNext()

Habe ich zwei observablen (mit Namen A und B der Einfachheit halber) und einen Abonnenten. So, die Abonnenten abonniert, und wenn es ist ein Fehler auf A dann B (das ist die fallback -) kicks in. Nun, wenn Eine trifft, einen Fehler B wird als gut, aber Ein Anruf, onComplete() auf die Abonnenten, also B-response-nie erreicht den Abonnenten selbst wenn B die Ausführung erfolgreich ist.

Ist das das normale Verhalten? Ich dachte onErrorResumeNext() sollte auch weiterhin den stream und unterrichtet den Teilnehmer nach Abschluss, wie schon in der Dokumentation (https://github.com/ReactiveX/RxJava/wiki/Error-Handling-Operators#onerrorresumenext).

Dies ist die Allgemeine Struktur dessen, was ich Tue (entfallen mehrere "langweilig" - code):

public Observable<ModelA> observeGetAPI(){
    return retrofitAPI.getObservableAPI1()
            .flatMap(observableApi1Response -> {
                ModelA model = new ModelA();

                model.setApi1Response(observableApi1Response);

                return retrofitAPI.getObservableAPI2()
                        .map(observableApi2Response -> {
                            //Blah blah blah...
                            return model;
                        })
                        .onErrorResumeNext(observeGetAPIFallback(model))
                        .subscribeOn(Schedulers.newThread())
            })
            .onErrorReturn(throwable -> {
                //Blah blah blah...
                return model;
            })
            .subscribeOn(Schedulers.newThread());
}

private Observable<ModelA> observeGetAPIFallback(ModelA model){
    return retrofitAPI.getObservableAPI3().map(observableApi3Response -> {
        //Blah blah blah...
        return model;
    }).onErrorReturn(throwable -> {
        //Blah blah blah...
        return model;
    })
    .subscribeOn(Schedulers.immediate());
}

Subscription subscription;
subscription = observeGetAPI.subscribe(ModelA -> {
    //IF THERE'S AN ERROR WE NEVER GET B RESPONSE HERE...
}, throwable ->{
    //WE NEVER GET HERE... onErrorResumeNext()
},
() -> { //IN CASE OF AN ERROR WE GET STRAIGHT HERE, MEANWHILE, B GETS EXECUTED }
);

Irgendwelche Ideen was ich falsch mache?

Dank!

EDIT:
Hier ist eine grobe timeline, was passiert ist:

---> HTTP GET REQUEST B
<--- HTTP 200 REQUEST B RESPONSE (SUCCESS)

---> HTTP GET REQUEST A
<--- HTTP 200 REQUEST A RESPONSE (FAILURE!)

---> HTTP GET FALLBACK A
** onComplete() called! ---> Subscriber never gets fallback response since onComplete() gets called before time.
<--- HTTP 200 FALLBACK A RESPONSE (SUCCESS)

Und hier ist ein link zu einem einfachen Diagramm habe ich gemacht, die darstellen, was ich will, geschehen:
Diagramm

  • Die timeline zeigt die HTTP 200 für die ausbleibende Antwort. Gibt es eine andere Möglichkeit Sie die Signalisierung eines Fehlers von getObservableAPI2()? Außerdem können Sie festlegen, welche API-Anfragen entspricht der timeline Ausgabe? Es sieht aus wie getObservableAPI1->REQUEST B, getObservableAPI2->fordern Sie EIN, getObservableAPI3->FALLBACK EIN, aber ich will nur sicher.
  • Ja, tatsächlich, obwohl die response ist eine 200 ein, einige Daten kommen kann, null ist, so dass ich werfen und Fehler in diesen Szenarien. Und ja, das ist die timeline-Anfragen Verhältnis, ich werde Bearbeiten Sie die Frage so schnell wie möglich zu entsprechen timeline-Anfrage, wie deine.
  • Ihre Logik sieht-sound. Sie sollten immer die fallback-Antwort, bevor onComplete. Können Sie entfernen Sie alle subscribeOn() aufrufen und sehen, was passiert. Sie sollte nicht erforderlich sein, da die Nachrüstung führt die Anforderungen auf seinen eigenen thread-pool sowieso.
  • Ich habe bereits versucht, dass und bekam die exakt gleiche Ausgabe, onComplete wird aufgerufen, zu früh.
  • Es ist besser, Sie zu glätten Ihre Kette statt der Schachtelung es (macht es super schwer zu Lesen, trace und debug). Es ist sehr unklar, was Sie zu tun versuchen hier insb. innerhalb der flatMap block. Bitte Räumen Sie Ihre Methoden und Variablen, ist es egal, ob es um die Nachrüstung oder nicht
InformationsquelleAutor mradzinski | 2014-09-03
  1. 6

    Den Rx-Aufrufe verwendet im folgenden soll simulieren, was Sie tun, mit Retrofit. 

    fallbackObservable =
        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        logger.v("emitting A Fallback");
                        subscriber.onNext("A Fallback");
                        subscriber.onCompleted();
                    }
                })
                .delay(1, TimeUnit.SECONDS)
                .onErrorReturn(new Func1<Throwable, String>() {
                    @Override
                    public String call(Throwable throwable) {
                        logger.v("emitting Fallback Error");
                        return "Fallback Error";
                    }
                })
                .subscribeOn(Schedulers.immediate());

stringObservable =
        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        logger.v("emitting B");
                        subscriber.onNext("B");
                        subscriber.onCompleted();
                    }
                })
                .delay(1, TimeUnit.SECONDS)
                .flatMap(new Func1<String, Observable<String>>() {
                    @Override
                    public Observable<String> call(String s) {
                        logger.v("flatMapping B");
                        return Observable
                                .create(new Observable.OnSubscribe<String>() {
                                    @Override
                                    public void call(Subscriber<? super String> subscriber) {
                                        logger.v("emitting A");
                                        subscriber.onNext("A");
                                        subscriber.onCompleted();
                                    }
                                })
                                .delay(1, TimeUnit.SECONDS)
                                .map(new Func1<String, String>() {
                                    @Override
                                    public String call(String s) {
                                        logger.v("A completes but contains invalid data - throwing error");
                                        throw new NotImplementedException("YUCK!");
                                    }
                                })
                                .onErrorResumeNext(fallbackObservable)
                                .subscribeOn(Schedulers.newThread());
                    }
                })
                .onErrorReturn(new Func1<Throwable, String>() {
                    @Override
                    public String call(Throwable throwable) {
                        logger.v("emitting Return Error");
                        return "Return Error";
                    }
                })
                .subscribeOn(Schedulers.newThread());

subscription = stringObservable.subscribe(
        new Action1<String>() {
            @Override
            public void call(String s) {
                logger.v("onNext " + s);
            }
        },
        new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                logger.v("onError");
            }
        },
        new Action0() {
            @Override
            public void call() {
                logger.v("onCompleted");
            }
        });

    Die Ausgabe aus dem log-Anweisungen ist:

    RxNewThreadScheduler-1 emitting B 
RxComputationThreadPool-1 flatMapping B 
RxNewThreadScheduler-2 emittiert Eine 
RxComputationThreadPool-2 A abgeschlossen, sondern enthält ungültige Daten - Fehler werfen 
RxComputationThreadPool-2-emitting-Fallback 
RxComputationThreadPool-1 onNext Fallback 
RxComputationThreadPool-1 onCompleted

    Dies scheint, wie das, was du suchst, aber vielleicht bin ich etwas fehlt.

    InformationsquelleAutor kjones
Schreibe einen Kommentar