RxJava 2 Ressourcen in umgekehrter Reihenfolge entsorgen

Question

Im Grunde, was ich hinterher bin, ist, wie Ressourcen 1,2,3,4 erstellen, Ergebnisse verarbeiten und dann 4,3,2,1 mit der Anforderung, dass, wenn es ist ein Fehler beim Erstellen der Ressource 3, wir müssen nur 2,1 entsorgen.

Als Beispiel lassen Sie uns sagen, dass wir dieses:

Observable1 liest Zeilen aus einer Datei und führt eine Emission pro Zeile. Für die richtige Handhabung der Dateiressource sollte die Datei am Anfang geöffnet und am Ende geschlossen werden.

Observable2 sollte die Emissionen von Observable1 akzeptieren und in eine andere Datei schreiben. Der Einfachheit halber wollen wir sagen, dass wir am Ende jeder Zeile nur ein Komma hinzufügen müssen. Für den richtigen Umgang mit einer anderen Datei-Ressource, sollten wir die neue Datei am Anfang erstellen und bündig und am Ende schließen.

ich denke, es so etwas wie

observable1.flatMap (line -> { doMagicHere(line+",") }) 
    .subscribe (line -> { System.out.println(line) }, 
    ex -> { ex.printStackTrace() } 
    ) 

aussehen sollte Wenn ein Fehler auftritt, während file1 öffnen, sollten wir nicht file2 erstellen (noch entsorgen) Wenn ein Fehler auftritt, während die Linien der Verarbeitung von file2 entsorgt werden zuerst und nur dann, sollte file1 entsorgt werden. (auch wenn es in diesem speziellen Beispiel so oder so funktioniert)

Ich denke, ich kann Observable1 mit dem Observable.using() - Operator erstellen, wo wir die Datei im ersten Parameter öffnen, das Observable im 2. Parameter erzeugen und, Schließlich, dispose (schließen) Sie die Datei in den 3D-Parameter.

Ich weiß nicht, wie man den "using" -Operator im zweiten Schritt anwendet, oder ob es der richtige Weg ist.

Vielen Dank für jede Lösung oder Zeiger (vorzugsweise in rxjava 2)

Answer 1

Sie haben einen guten Start mit Observable.using(). Allerdings ist die erste beobachtbar ist:

Observable<Observable<String>> observable1; 

Durch diese Umleitung hinzufügen, können Sie mit der Situation, wo die erste Ressource geladen werden kann.

interface FileWriter { 
    void write(String line); 
} 
Observable<FileWriter> makeDestinationFile(String filename); 

Durch die FileWriter Abstraktion Einführung, können Sie dann die zweite Ressource binden.

observable1.flatMap(
    lineSource -> 
     Observable.combineLatest(lineSource, 
     makeDestinationFile(filename), 
     (line, writer) -> writer.write(line + ",")) 
    .subscribe(); 

Wenn observable1 Abschluss erreicht, werden beide Observablen abgeschlossen und die zugrunde liegenden Ressourcen freigegeben.