node.js - 将三个不同的函数映射到 Node.js 中的 Observable

Question

我是 Rxjs 新手。如果可能的话，我想遵循最佳实践。

我试图对可观察对象中返回的相同数据执行三个不同的函数。遵循“数据流”的概念，我一直认为我需要将这个 Observable 分成三个流并继续。

这是我的代码，所以我可以停止抽象地谈论:

// NotEmptyResponse splits the stream in 2 to account based on whether I get an empty observable back.  
let base_subscription = RxNode.fromStream(siteStream).partition(NotEmptyResponse);

// Success Stream to perform further actions upon. 
let successStream = base_subscription[0];

// The Empty stream for error reporting
let failureStream = base_subscription[1];

//Code works up until this point. I don't know how to split to 3 different streams. 
successStream.filter(isSite)
             .map(grabData)// Async action that returns data
             /*** Perform 3 separate actions upon data that .map(grabData) returned **/ 
             .subscribe(); 

如何将此数据流分成三个，并将每个数据实例映射到不同的函数？

Answer 1

事实上 partition() 运算符内部只是 calls filter() operator twice 。首先根据与谓词匹配的值创建一个Observable，然后根据与谓词不匹配的值创建一个Observable。

所以你可以用 filter() operator 做同样的事情:

let obs1 = base_subscription.filter(val => predicate1);
let obs2 = base_subscription.filter(val => predicate2);
let obs3 = base_subscription.filter(val => predicate3);

现在你有了三个 Observables，每个 Observables 只发出一些特定的值。然后您可以继续使用现有代码:

obs2.filter(isSite)
    .map(grabData)
    .subscribe();

请注意，调用 subscribe() 会触发源 Observable 生成值。这并不一定总是这样，具体取决于您使用的 Observable。请参阅“Hot” and “Cold” Observables在文档中。 Operator connect()根据您的用例，可能对您有用。