reactive-programming - 如何合并两个流(不包含null)并在对上应用条件？

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)....

我刚刚开始使用响应式(Reactive)流的想法，所以如果我误解了响应式(Reactive)流的想法，请纠正我。

Answer 1

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

Zip :

RxJ的引用:https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/zip.md

CombineLatest :

RxJ的引用:https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/combinelatest.md

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

不幸的是，两个运算符(operator)都无法获得您描述的这种行为:

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将取决于事件在Streams中发生的顺序，当然，您将把 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).... 

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

有多种方法构成可观察对象，其他未提及的运算符是:

可以在组合的最后添加

distinctUntilChanged，使用键选择器功能来限制部分邮政编码或最新值。

switch，用于将一个可观察到的内部组合在一起。