如何聚合多个 JavaRX Observables

Question

我有一个 Observable 列表，我想知道它们何时全部完成。特别是我想知道它们是否全部成功完成或其中任何一个有错误。列表中的所有可观察对象都应在成功或错误通知执行之前完成。我不需要有关这些通知中的结果或错误的任何信息。

我尝试了 Observable.merge(list)，但它因第一个错误而停止。

想要的示例

Observable.all(listOfObservables)
    .onSuccess(() -> {}) // only called if all completed successfully
    .onError(() -> {}) // only called after all completed and at least one errored
    .onComplete(() -> {}; // called after all completed regardless of success or error

这似乎是一种标准行为；其他 'async promise' 之类的框架支持这一点。 JavaRx 是否支持这种行为？实现此目标的最佳方法是什么？

Answer 1

我认为你需要 mergeDelayError。它合并多个 Observable，与 merge 相同，但它的 JavaDoc 说 - 将 N Observable 扁平化为一个 Observable，以一种允许 Observer 从中接收所有成功发射的项目的方式所有源 Observables 都不会被其中之一的错误通知打断。因此，源 Observable 发出的错误不会终止产品 Observable。所以：

Observable.mergeDelayError(listOfObservables)
  .subscribe(
    (v) -> {}, // for every successfull item
    (e) -> {}, // after all completed and at least one errored
    () -> {}  // after all completed, won't be called on error though
  );

如何聚合多个 JavaRX Observables

How to aggregate multiple JavaRX Observables

java

rx-java