在使用流之前做一些事情，使用highland.js

Question

我试图编写一个可写流，它接受对象流并将它们输入到mongodb数据库中。在使用对象流之前，我首先需要等待数据库连接的建立，但我似乎做错了什么，因为程序永远不会到达插入部分。

// ./mongowriter.js

let mongo = mongodb.MongoClient,
    connectToDb = _.wrapCallback(mongo.connect);

export default url => _.pipeline(s => {
  return connectToDb(url).flatMap(db => {
    console.log('Connection established!');
    return s.flatMap(x => /* insert x into db */);
  });
});

....

// Usage in other file
import mongowriter from './mongowriter.js';

let objStream = _([/* json objects */]);

objStream.pipe(mongoWriter);

该程序刚刚退出，没有“连接建立！”被写到控制台上。

，我错过了什么？有什么成语我应该听吗？

Answer 1

通过阅读源代码和一些一般的实验，我了解了如何做一个异步的事情，然后继续在流中进行处理。基本上，使用flatMap将异步任务中的事件替换为实际要处理的流。

我没想到的另一个怪癖是，除非在回调中完全消耗了原始流，否则_.pipeline将无法工作。这就是为什么它不能简单地输入一个_.map和日志(这就是我试着调试它的方式)的原因。相反，我们需要确保在最后有一个each或done。下面是一个很小的例子：

export default _ => _.pipeline( stream => {
  return _(promiseReturningFunction())
    .tap(_ => process.stdout.write('.'))
    .flatMap(_ => stream)
    .each(_ => process.stdout.write('-'));
});

// Will produce something like the following when called with a non-empty stream.
// Note the lone '.' in the beginning.
// => .-------------------

基本上，a '.‘’在完成异步函数时输出，对于流的每个对象输出“-”。

希望这能节省一些时间。我花了很长时间才弄明白这件事。^^