RxJava: Erstellen Sie einen eigenständigen Stream

Question

Sagen, ich möchte eine Observable in RxJava machen, die eine Feedback-Kopplung wie auf dem Bild unten hat.

Ich habe es geschafft, dass bei der Verwendung von Probanden zu erreichen, wie folgt aus:

// Observable<Integer> source = Observable.range(0, 6); 

public Observable<Integer> getFeedbackSum(Observable<Integer> source) { 
    UnicastSubject<Integer> feedback = UnicastSubject.create(); 
    Observable<Integer> feedbackSum = Observable.zip(source, feedback.startWith(0), Pair::create) 
     .map(pair -> pair.first + pair.second); 

    feedbackSum.subscribe(feedback); 
    return feedbackSum; 
} 

Es ziemlich hässlich aussieht. Gibt es einen besseren Weg?

Answer 1

Es ist ein Betreiber dafür: scan:

public final <R> Observable<R> scan(R initialValue, BiFunction<R,? super T,R> accumulator)

Observable.range(0, 6) 
    .scan(0, (a, b) -> a + b) 
    .test() 
    .assertResut(0, 1, 3, 6, 10, 15, 21); 

Wenn Ihr Akku-Typ nicht unveränderlich ist, können Sie scanWith:

public final <R> Observable<R> scanWith(Callable<R> seedSupplier, BiFunction<R,? super T,R> accumulator)