RxJava Observable combineLatest 重置状态

Question

基于官方文档的漂亮图表，combineLatest http://reactivex.io/documentation/operators/combinelatest.html

我想知道是否有办法，比如当我从第二个 Observable 收到“C”时，我想清除第一个和第二个 Observable 内部的状态。这样当“D”从第二个 Observable 到达时，它将不再与来自第一个 Observable 的“2”结合，而是等到“3”到达。

Answer 1

解决您的确切问题的方法是发布每个源，直到触发值，然后重新订阅您的链到源：

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");

这样，您可以终止组合的第一个“会话”并开始一个新的。