如何使用Highland.js编写过滤器(使用DB )

Question

我正在尝试使用Highland.js设计一个工作流。我还不能弄清楚Highland.js是如何使用它的。

我有一个基于流的工作流程，如下(伪代码)，

read                      //fs.createReadStream(...)
   .pipe(parse)           //JSONStream.parse(...)
   .pipe(filterDuplicate) //mongoClient.db.collection.count({}) > 0
   .pipe(transform)       //fn(item) { return tranform(item); }
   .pipe(write);          //mongoClient.db.collection.insert(doc)

filterDuplicate查找数据库以检查读取的记录是否存在(使用一个条件)，并返回一个布尔结果。为了让过滤器工作，它需要一个活动的DB连接，我希望在流完成之前重复使用它。一种方法是在写入的“完成”事件之前打开连接，这意味着我需要将连接作为过滤和写入的参数传递，如果这两种方法使用相同的数据库，这将起作用。

在上面的工作流程中，filterDuplicate和write也可以使用不同的数据库。因此，我希望在每个函数中包含和管理连接，这使得它成为一个自包含的可重用单元。

我正在寻找任何关于如何使用高地设计这一点的意见。

谢谢。

Answer 1

这不会像只使用一堆pipe那么简单。您必须为该任务使用最合适的API方法。

这是一个粗略的例子，展示了你可能会接近的结果：

read
  .through(JSONStream.parse([true]))
  .through((x) => {
    h((next, push) => { // use a generator for async operations
      h.wrapCallback( mongoCountQuery )( params ) // you don't have to do it this way
        .collect()
        .toCallback((err, result) => {
          if ( result > 0 ) push( err, x ); // if it met the criteria, hold onto it
          return push( null, h.nil ); // tell highland this stream is done
        });
    });
  })
  .merge() // because you've got a stream of streams after that `through`
  .map(transform) // just your standard map through a transform
  .through((x) => {
    h((next, push) => { // another generator for async operations
      h.wrapCallback( mongoUpdateQuery )( params )
        .toCallback((err, results) => {
          push( err, results );
          return push( null, h.nil );
        });
    });
  })
  .merge() // another stream-of-streams situation
  .toCallback( cb ); // call home to say we're done