如何在 node.js 中编写转换流答案

【问题标题】：How to compose transform Streams in node.js如何在 node.js 中编写转换流
【发布时间】：2014-07-05 01:22:26
【问题描述】：

我有一个实现为一系列转换流的 csv 解析器：

process.stdin
    .pipe(iconv.decodeStream('win1252'))
    .pipe(csv.parse())
    .pipe(buildObject())
    .pipe(process.stdout);

我想抽象解析器（在它自己的模块中）并且能够做到：

process.stdin.
    .pipe(parser)
    .pipe(process.stdout);

其中parser 只是之前使用的转换流的组合。

如果我这样做了

var parser = iconv.decodeStream('win1252')
    .pipe(csv.parse())
    .pipe(buildObject());

然后将parser 设置为 buildObject() 流，并且只有此转换流接收数据。

如果我这样做了

var parser = iconv.decodeStream('win1252');
parser
    .pipe(csv.parse())
    .pipe(buildObject());

它也不起作用，因为.pipe(process.stdout) 将在第一个转换流上被调用，而另外两个将被绕过。

对优雅的流组合有什么建议吗？

【问题讨论】：

标签： node.js stream

【解决方案1】：

不幸的是，没有内置的方法可以做到这一点，但是有很酷的multipipe 包。像这样使用：

var multipipe = require('multipipe');

var parser = multipipe(iconv.decodeStream('win1252'), csv.parse(), buildObject());

【讨论】：

相同。这太棒了。此外，multipipe 的源代码非常短，因此值得任何重视底层了解的人查看其实现。
还有pumpify，同样可以让你组合多个流。

【解决方案2】：

我一直在努力解决这个问题（以及其他一些问题！）。我发现highlandjs 几乎解决了我所有的问题。在这种情况下，他们的pipeline 命令成功了：

var h = require('highland');
var parser = h.pipeline(iconv.decodeStream('win1252'), csv.parse(), buildObject());

【讨论】：

我会为了这个答案卖掉我的第一个孩子。谢谢！

【解决方案3】：

我认为现在可以在本地完成。

const { PassThrough, Transform } = require('stream');

const compose = (...streams) => {
  const first = new PassThrough();
  const last = new PassThrough();
  const result = new Transform();

  [first, ...streams, last].reduce(
    (chain, stream) => (
      stream.on('error', (error) => result.emit('error', error)),
      chain.pipe(stream)
    ),
  );

  result._transform = (chunk, enc, cb) => {
    last.once('data', (chunk) => cb(null, chunk));
    first.push(chunk, enc);
  };

  result._flush = (cb) => {
    last.once('end', () => cb(null));
    first.push(null);
  };

  return result;
};

【讨论】：