rxjava 区间与另一个 observable 相结合

Question

我有一个通过系统进程执行操作的 observable（很热），并且我希望也有一个间隔运行，直到进程 observable 命中 onComplete。

我看到区间运算符：http://reactivex.io/documentation/operators/interval.html

我怎样才能将这两者结合起来以获得我想要的行为（特别是取消其他点击 onComplete 时的间隔）？

Answer 1

您可以使用takeUntil() 运算符来取消（取消订阅）区间运算符，takeUntil() 将Observable 作为输入，并在输入Observable 发出项目时取消。

剩下的就是将热的Observable 转换为Observable，它发出带有onComplete() 的项目，因此我们可以将其用作takeUntil() 的输入，这可以使用materialize() 运算符，为每个Observable 事件（onNext()、onError()、onCompleted()）发出Notification 对象，与filter() 一起仅接收onCompleted() 事件。

Observable<Notification<Object>> hotOnCompleteObservable = 
            hot.materialize()
               .filter(notification -> notification.isOnCompleted());

Observable interval = ...
interval.takeUntil(hotOnCompleteObservable);