RxJava onErrorResumeNext()

J'ai deux observables (nommées A et B pour des raisons de simplicité) et un abonné. Ainsi, l'Abonné souscrit à Un et si il y a une erreur sur A puis B (qui est la solution de repli) des coups de pied dans. Maintenant, chaque fois qu'Une rencontre une erreur B est appelé bien, mais les appels de onComplete() sur l'abonné, B réponse n'a jamais atteint l'abonné, même si l'exécution est réussie.

Est-ce le comportement normal? J'ai pensé onErrorResumeNext() devrait continuer à la diffuser et informer l'abonné une fois rempli comme indiqué dans la documentation (https://github.com/ReactiveX/RxJava/wiki/Error-Handling-Operators#onerrorresumenext).

C'est la structure globale de ce que je fais (omis plusieurs "ennuyeux" 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 }
);

Les idées de ce que je fais mal?

Merci!

EDIT:
Voici une ébauche de calendrier de ce qui se passe:

---> 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)

Et voici un lien vers un schéma simple que j'ai faite qui représente ce que je veux:
Diagramme

Votre timeline montre HTTP 200 pour la réponse en cas d'échec. Est-il une autre manière de signalisation d'une erreur de getObservableAPI2()? Aussi, pouvez-vous spécifier les requêtes à l'API correspond à la chronologie de sortie? Il ressemble à getObservableAPI1->DEMANDE de B, getObservableAPI2->DEMANDER UN, getObservableAPI3->SECOURS d'UNE, mais je veux juste m'en assurer.
Oui, en fait, bien que la réponse est un 200, certaines données peuvent venir null, donc je l'ai lancer et d'erreur dans ces scénarios. Et oui, c'est le scénario-les demandes de rapport, je vais modifier la question le plus tôt possible pour correspondre à la timeline-demande que le vôtre.
Votre logique semble sain. Vous devriez obtenir le secours de réponse avant onComplete. Pouvez-vous supprimer tous les subscribeOn() appels et voir ce qui se passe. Ils ne devraient pas être nécessaire puisque la Rénovation exécute les demandes sur son propre pool de threads de toute façon.
J'ai déjà essayé, et a obtenu exactement la même sortie, onComplete est appelé trop tôt.
C'est mieux pour aplatir votre chaîne au lieu d'imbrication (il est super dur à lire, de suivi et de débogage). Il est très difficile de ce que vous essayez de faire ici de l'esp. dans le flatMap bloc. Veuillez ranger vos méthodes et de variables, il n'a pas le temps il Rétrofit ou pas

InformationsquelleAutor mradzinski | 2014-09-03

Le Rx appels utilisés dans les éléments suivants doivent simuler ce que vous faites avec de Rénovation.

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

De la sortie du journal des états est:

RxNewThreadScheduler-1 émettant B 
RxComputationThreadPool-1 flatMapping B 
RxNewThreadScheduler-2 émettant Un 
RxComputationThreadPool-2 se termine, mais contient des données non valides des jets d'erreur 
RxComputationThreadPool-2 émettant Un de Secours 
RxComputationThreadPool-1 onNext de Secours 
RxComputationThreadPool-1 onCompleted

Cela semble être ce que vous cherchez, mais peut-être que je suis absent quelque chose.

InformationsquelleAutor kjones

Vous devez vous connecter pour publier un commentaire.