如何合并两个流（没有空值）并在对上应用条件？

Question

假设我有两个数据流，有没有办法合并它们并对这两个流之间的数据应用条件？例如

Stream A : A, B, C, D....
Stream B : -, A, -, -....
Composed : (A,-),(B,A),(C,-),(D,-)....

如何使用rxjs获取上面的组合流？我想对组合流应用条件以引发一些通知。也可以使用最后已知的非空数据，例如参见下面的组合流。

Stream A : A, B, C, D....
Stream B : 1, null, 2, null....
Composed : (A,1),(B,1),(C,2),(D,2)....

我刚刚开始玩反应流的想法，所以如果我误解了反应流的想法，请纠正我。

Answer 1

有两个运营商可以为您服务。

Zip:

RxJs 参考：https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/zip.md

CombineLatest:

RxJs 参考资料：https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/combinelatest.md

图像解释了两者之间的差异。现在您已经合并了您只需要过滤的可观察对象，使用 where，如果其中一个值为 null，它将进行过滤。

不幸的是，这两个运营商都无法获得您描述的这种行为：

Stream A : A, B, C, D, E....
Stream B : 1, null, 2, null, 3....
Composed : (A,1),(B,1),(C,2),(D,2)....

如果使用 Zip 和 Where（之后过滤空值），结果将是：

Composed: (A,1),(C,2),(E,3)

如果使用 Where（之前过滤空值）和 Zip，结果将是：

Composed: (A,1),(B,2),(C,3)

如果您使用 CombineLatest 将取决于事件在流中发生的顺序，当然，您将放置 where 运算符的位置，结果可能与您显示的不同，例如：

Stream A : A, B, C, D....
Stream B : 1, null, 2, null....
Composed : (A,1),(B,1),(C,1),(C,2),(D,2).... // OR
Composed : (A,1),(B,1),(B,2),(C,2),(D,2)....

除非您有更具体的要求，否则我认为我提到的选项之一就是您正在寻找的，请随时添加信息。

有几种方法可以组成 observable 的，其他未提及的运算符是：

distinctUntilChanged，可以添加在合成的最后，使用key selector函数来限制为zip的一部分或最新值。
switch，用于将一个 observable 组合到另一个中。

如何合并两个流（没有空值）并在对上应用条件？

How to merge two streams (without nulls) and apply conditions on pairs?

reactive-programming

rxjs

reactive-streams