Rxjs 5:如何建立观察链？

Question

我是RxJS的新手，我正在尝试实现一系列行为完全像MS-Excel的可观察对象。概念:假设excel有5列'Name'，'Age'，'Sex'，'Country'，'Zipcode‘。我们可以对每一列单独应用过滤器，这也会影响其他列中显示的记录。

这里数据源接收数据后端服务，数据源将只有两个函数"addRecord“和"removeRecord”。

我试图在这里实现的方式，假设我将创建可观察对象并附加到数据源，将其称为OBS-1，这将从数据源接收数据。OBS-1可以有自己的过滤器。假设我将创建另一个可观察的OBS-2，它将接收数据OBS-1 (如果OBS-1中有任何过滤器，则过滤数据)。另一个观察点是OBS-3，它再次从OBS-2接收数据(如果在OBS-2中有数据，则进行过滤)，依此类推。

如果OBS-2被销毁(取消订阅)，则OBS-3将从OBS-1接收数据。

我们如何在RxJ中实现这一点？

Answer 1

我想你误解了关于Rx的一些事情。可观察对象没有过滤器，您也不能在其中‘实时’添加和删除过滤器。观察者也不会根据订阅的人转发数据。

相反，您需要构建一个调用链。您可以从一个源可观察对象开始，比如一个来自addRecord，一个来自removeRecord事件。然后，通过Rx中的各种operators，将这些可观测对象链接起来，形成新的可观测对象，最终订阅最终的可观测对象。订阅将激活整个链，当源事件触发时，所有操作符都将触发，最终事件将(如果未过滤)到达subscribe。

你实际上可以做你在Rx中描述的事情。例如，使用switchMap可以相对容易地更改可观察对象上的过滤器，这是一种操作符，允许您将一个序列投影到另一个序列上，并每次都切换到新序列。例如filterSource.switchMap(filterFunction => Obs-1.filter(filterFunction))。甚至比这更简单的是，您可以取消订阅第一个订阅，并再次设置Rx链。但是，使用内置函数会将许多杂耍状态排除在等式之外。

然而，我强烈怀疑您实际上并不需要如此复杂的行为。您想要的内容可以像这样简单地归档：

var Src-1 = fromEvent(dataSource, 'addRecord') // create the first source
var Src-2 = fromEvent(dataSource, 'removeRecord') // and the other source
var Obs-1 = Src-1.combineLatest(Src-2) // combine both sources
  .filter(e => someCondition(e)) // filter the source
var Obs-2 = Obs-1.mergeMap(e => someOtherCondition(e) ? Change(e) : Rx.Observable.of(e)) // on someOtherCondition, either transform the source with the `Change(e)` function. Or keep it unchanged with `of(e)`
var Obs-3 = Obs-2.filter(e => anotherCondition(e)) // Filter again
var sub = Obs-3.subscribe() // activate the sequence.