rxjs - RxJava/RxJs : How to merge two source observables but complete as soon as one of them completes

Question

我有两个源可观察对象。
我想合并两个源 observables，但是只要其中一个源 observables 完成，合并的 observable 就会完成。

期望的行为:

Source 1: ---1--------3--4-----------------------------x
Source 2: -------2----------x
"merged"  ---1---2----3--4--x

如果其中一个源出现错误，则该错误应传播到合并的 observable:

Source 1: ---1--------3--4-----------------------------x
Source 2: -------2----------e
"merged"  ---1---2----3--4--ex

“合并”运算符仅在两个源都完成后才完成合并流:

Source 1: ---1--------3--4-----------------------------x
Source 2: -------2----------x
"merged"  ---1---2----3--4-----------------------------x

我怎样才能实现我想要的行为？

Answer 1

您需要使用元数据，每个 observable 的信息。为此，请使用 materialize()每个流上的运算符和使用 dematerialize()在合并的流上实际发出数据。

Observable.merge( observableA.materialize(),
                  observableB.materialize() )
  .takeWhile( notification -> notification.hasValue() )
  .dematerialize()
  .subscribe( ... );

这将合并两个 observable，直到其中一个完成或发出错误。