Wie wird in RxJava eine Variable beim Verketten von Observablen weitergegeben?

Ich verkette asynchrone Operationen mit RxJava und möchte eine Variable nachgeschaltet übergeben:

Observable
   .from(modifications)
   .flatmap( (data1) -> { return op1(data1); })
   ...
   .flatmap( (data2) -> { 
       // How to access data1 here ?
       return op2(data2);
   })

Es scheint ein allgemeines Muster zu sein, aber ich konnte keine Informationen darüber finden.

Der Rat, den ich vom Couchbase-Forum erhalten habe, ist die Verwendung verschachtelter Observablen:

Observable
   .from(modifications)
   .flatmap( (data1) -> { 
       return op1(data1)
           ...
           .flatmap( (data2) -> { 
               // I can access data1 here
               return op2(data2);
           })
   });

EDIT: Ich werde dies als akzeptierte Antwort markieren, da es am meisten empfohlen wird. Wenn Ihre Verarbeitung zu komplex ist, um alles zu verschachteln, können Sie die Lösung auch mit Funktionsaufrufen überprüfen.

Eine andere Möglichkeit besteht darin, das Ergebnis von a op1auf eine org.apache.commons.lang3.tuple.PairVariable abzubilden , die die Variable enthält, und diese weiterzugeben:

Observable
   .from(modifications)
   .flatmap( (data1) -> {
       return op1(data1).map( obj -> { return Pair.of(data1,obj); });
   })
   ...
   .flatmap( (dataPair) -> { 
       // data1 is dataPair.getLeft()
       return op2(dataPair.getRight());
   })

Es funktioniert, aber es ist etwas unangenehm, Variablen in einem Pair / Triple / ... versteckt zu haben, und es wird sehr ausführlich, wenn Sie die Java 6-Notation verwenden.

Ich frage mich, ob es eine bessere Lösung gibt, vielleicht könnte ein RxJava-Operator helfen?

Flatmap kann ein zweites Argument dauern:

Observable.just("foo")
                .flatMap(foo -> Observable.range(1, 5), Pair::of)
                .subscribe(pair -> System.out.println("Result: " + pair.getFirst() + " Foo: " + pair.getSecond()));

Quelle: https://medium.com/rxjava-tidbits/rxjava-tidbits-1-use-flatmap-and-retain-original-source-value-4ec6a2de52d4

Eine Möglichkeit wäre, einen Funktionsaufruf zu verwenden:

private static Observable<T> myFunc(final Object data1) {
    return op1(data1)
        ...
        .flatmap( (data2) -> { 
            // I can access data1 here
            return op2(data2);
        });
}

Observable
   .from(modifications)
   .flatmap( (data1) -> { return myFunc(data1); })

ABER: Korrigieren Sie mich, wenn ich falsch liege, aber es fühlt sich nicht nach reaktiver Programmierung an

Eigentlich haben wir eine Bibliothek, die Anrufketten vereinfacht.

https://github.com/pakoito/Komprehensions

Hinzufügen als Gradle-Abhängigkeit:

implementation 'io.reactivex.rxjava2:rxjava:2.2.1'
implementation 'com.github.pakoito.Komprehensions:komprehensions-rx2:1.3.2'

Verwendung (Kotlin):

val observable = doFlatMap(
    { Observable.from(modifications) },
    { data1 -> op1(data1) },
    { data1, data2 -> op2(data2) },
    { data1, data2, data3 -> op3(data1, data2, data3) }
)

Die Lösung für diesen Thread funktioniert, aber für komplexe Ketten ist das Lesen von Code schwierig. Ich musste mehrere Werte übergeben. Ich habe eine private Klasse mit allen Parametern erstellt. Ich finde, dass Code auf diese Weise besser lesbar ist.

private class CommonData{
   private string data1;
   private string data2;

   *getters and setters*
}
...
final CommonData data = new CommonData();
Observable
   .from(modifications)
   .flatmap( (data1) -> { 
       data.setData1(data1);
       return op1(data1); 
   })
   ...
   .flatmap( (data2) -> { 
       data2 = data.getData1() + "data 2... ";
       data.setData2(data2);
       return op2(data2);
   })

ich hoffe es hilft

Ich weiß, dass dies eine alte Frage ist, aber mit RxJava2 & Lambda können Sie Folgendes verwenden:

Observable
.from(modifications)
.flatMap((Function<Data1, ObservableSource<Data2>>) data1 -> {
                        //Get data 2 obeservable

                            return Observable.just(new Data2())
                        }
                    }, Pair::of)

Beim nächsten Fluss (Flatmap / Map) lautet Ihr Ausgabepaar (Daten1, Daten2).

Sie können die "globale" Variable verwenden, um dies zu erreichen:

 Object[] data1Wrapper = new Object[]{null};
 Object[] data2Wrapper = new Object[]{null};
 Observable
    .from(modifications)
    .flatmap(data1 -> {
        data1Wrapper[0] = data1;
        return op1(data1)
     })
      ...
    .flatmap(data2 -> { 
        // I can access data1 here use data1Wrapper[0]
        Object data1 = data1Wrapper[0];
        data2Wrapper[0] = data2;
        return op2(data2);
     })