Async-Bereinigung bei beobachtbarer Sequenzbeendigung

Ich muss eine asynchrone Methode (z. B. einen Bereinigungsjob) bei einem beobachtbaren Abschluss / Fehler ausführen. Auch wenn die Bereinigungsmethode fehlschlägt, sollte die beobachtbare Kette ebenfalls fehlschlagen.

Gibt es eine Standardmethode dafür?

Angenommen, ich habe eine Sequenz Observable source und eine asyncCleanup () -Methode, die ein beobachtbares Bereinigungsergebnis zurückgeben.

Die Nebenwirkungsmethoden wie doOnCompleted / doOnTerminate / doOnUnsubscribe scheinen nicht geeignet zu sein:

source.doOnUnsubscribe(()->asyncCleanup().subscribe());

Die beobachtbare Kette ist auch dann erfolgreich, wenn asyncCleanup () fehlschlägt. AsyncCleanup () sollte also Teil derselben Kette sein.

Das Beste, was ich mir ausgedacht habe, ist das Folgende:

source.onErrorResumeNext(err->asyncCleanup()
.flatMap(cleanupResult->Observable.error(err)))
.concatWith(asyncCleanup()
.flatMap(cleanupResult->Observable.empty()));

Im Fehlerfall ruft onErrorResumeNext aufasyncCleanup () und wird dem ursprünglichen Fehler zugeordnet. Im Erfolgsfall können wir mit asyncCleanup () verknüpfen, das einer leeren Sequenz zugeordnet ist. Leider funktioniert es nicht, wenn ein take () - oder ähnlicher einschränkender Operator nachgeschaltet ist, der dazu führen kann, dass das verkettete Observable nicht einmal abonniert wird.

UPDATE 01.08.2017: Die Frage bezieht sich speziell auf Sequenzobservablen. Für eine beobachtbare Einzelelementquelle ist die Lösung ganz einfach:

singleItemSource
.onErrorResumeNext(err->asyncCleanup()
.flatMap(cleanupResult->Observable.error(err)))
.flatMap(single->asyncCleanup()
.map(cleanupResult->single));

Antworten:

Die Idee, die Sie vorschlagen, bricht die Observable Vertrag, weil:

Rx-Designrichtlinien, 6.8 - Abmeldung sollte nicht werfen</ em>

public Observable<Long> asyncCleanup() {
...
return Observable.empty();
}

//Concat asyncCleanup() to be executed after the sequence
Observable.rangeLong(1, 10)
.concatWith(asyncCleanup())
.subscribe(System.out::println, System.out::println);

Observable.range(1, 10)
.flatMap(Observable::just, e -> asyncCleanup(), () -> asyncCleanup())
.subscribe(...);

Observable<Long> range = Observable.rangeLong(1, 10);

Observable.using(() -> range, //The resource provider
observable -> observable, //The resource
observable -> asyncCleanup() //The resource disposer
.subscribe(System.out::println, System.out::println))
.subscribe(System.out::println, System.out::println);