java - 如何检测 Observable<Observable<Object>> 中所有包含的 Observable 的完成

Question

我希望我的 api 上有一个方法返回 Observable> 但我希望该方法中的代码知道所有包含的 Observables 是否已完成，以便它可以关闭某些内容。最好的方法是什么？

更明确地说，我希望完成此方法:

public static <T> Observable<Observable<T>> doWhenAllComplete(
        final Observable<Observable<T>> original, Action0 action) {
  ...
}

Answer 1

抱歉，我的答案是在.NET中(就像system.reactive标签一样)；我相信你可以翻译它!

如果您的IObservable<IObservable<Object>>由 source 给出那么:

source.Merge()
      .Subscribe(_  => {}, /* not interested in onNext */
                 () => /* onCompleted action here, called when all complete */);

注意:如果任何流出错(导致合并流在此时终止)，这将失败，因此您也可以执行此操作来吞掉各个流上的错误:

source.SelectMany(x => x.Catch(Observable.Empty<Object>()))
      .Subscribe(_  => {}, /* not interested in onNext */
                 () => /* onCompleted action here, called when all complete */);