RxJava Observable combineLatest reset state

Question

Based on the nice diagram from official documentation, combineLatest http://reactivex.io/documentation/operators/combinelatest.html

I wonder if there is a way, say when I received "C" from the second Observable, I would like to clear the states inside both first and second Observables. So that when "D" arrives from second Observable, it will no longer combine with "2" from the first Observable but wait until "3" arrives.

akarnokd akarnokd · Accepted Answer · 2015-07-08T18:24:39

A solution to your exact problem is to publish each source, take until the trigger value and then resubscribe your chain to the sources:

PublishSubject<Integer> a = PublishSubject.create();
PublishSubject<String> b = PublishSubject.create();

ConnectableObservable<Integer> ap = a.publish();
ConnectableObservable<String> bp = b.publish();
Observable.combineLatest(
        ap, bp,
        Pair::of
)
.takeUntil(p -> "C".equals(p.second))
.repeat(2)
.subscribe(System.out::println)
;

ap.connect();
bp.connect();

a.onNext(1);
b.onNext("A");
a.onNext(2);
b.onNext("B");
b.onNext("C");

a.onNext(3);
b.onNext("D");

This way, you terminate the first 'session' of combination and start a new one.

RxJava Observable combineLatest reset state

1 Answers