将异步转换为Rx.js

Question

因此，我们正在尝试将我们的express服务器重写为Rx。它目前正在对所有流操作使用async。代码如下所示：

var async = require('async');

function getCountAndChannels(name, cb){
    var tasks = [
        function(cb) {
             //does a mongoDB search and returns count
        },
        function(cb) {
            //does a findOne mongoDB search and returns 
        }
    ];
    async.parallel(tasks, cb);
}

router.get('data', function(req, res) { //router is the express router
    var recorders = req.query.recorders.split(',');

    async.map(recorders, function(name, cb) {
        getCountAndChannels(name, cb);
    }, function(err, countsAndChannels) {
        if(err) throw err;

        // here countsAndChannels is an array with first element the count
        // and second element the document.

        // do other async stuff based on the results

        res.status(200).json('send some calculations');
});

这里我要做的是循环遍历recorders数组，并为每个数组计算两次mongoDB搜索。我尝试过使用Rx.Observable.merge，它不是以数组的形式返回结果，而是在两次不同的回调调用中返回结果。所以，然后我尝试了Rx.Observable.zip，我相信这就是我要找的。

问题是我不知道如何遍历recorders并在所有操作完成后发送结果。因为一个简单的forEach循环将抛出一个Cannot set headers after they are sent错误。

这就是我到目前为止所知道的：

recorders.forEach(recorder => {        
    Rx.Observable.zip([
        search1,
        search2
    ]).subscribe(
        (countsAndChannels) => {                
            // do stuff
            res.send('the results');
        },
        err => res.status(500).json(err),
        () => res.send('OK')
    );  
});

我以前没有使用过Rx，所以如果有任何帮助，我们将不胜感激。

Answer 1

将记录器列表转换为可观察的流，然后对每个记录器执行flatMap (即执行异步处理)，然后调用toArray将所有结果存储到一个数组中，这可能会更容易：

var recorder$ = Rx.Observable.from(recorders);
var countsAndChannels$ = recorder$
    .flatMap(performAsyncTask);

// allResults$ will emit once all of the async work is complete
var allResults$= countsAndChannels$.toArray();

allResults$.subscribe(results => {
    // Send response to client;
});