asynchronous - 如何使用异步功能消费高原流？

Question

使用异步函数消费高地流的正确模式是什么？似乎 .each 方法不适用于使用 wrapCallback 包装的节点样式异步函数。

我想做类似以下的事情。请注意，我还想捕获 .each 函数中生成的错误，并在所有异步消耗完成后触发 .done 函数。

function asyncConsume(item,cb) {
  // perform async operation based on data stream here
  return cb(null,item);
}
var wrappedFunction = _.wrapCallback(asyncConsume);

highlandStream.each(wrappedFunction).done('All successfully done!')
  .stopOnError(function(err) {
    console.log('This error handler catches errors in .each() as well.');
    });

谢谢你。

Answer 1

据我所知，您只能将异步函数与 Highlandmap函数一起使用（在函数被wrapCallback编辑之后）。地图实际上并没有调用该函数，因此您需要一个 Highlandseries或parallel调用来跟随并实际处理调用。

Answer 2

我认为您想要flatMap在这里，因为您正在将您的项目转换为一个新的流：

highlandStream
  .flatMap(function(item) {
    return wrappedFunction(item);
}).each(someFn).done('All successfully done!')
  .stopOnError(function(err) {
    console.log('This error handler catches errors in .each() as well.');
    });