节点：在可写流中接收 JSON答案

【问题标题】：Node: Receive JSON in a writable stream节点：在可写流中接收 JSON
【发布时间】：2015-06-28 20:15:21
【问题描述】：

我有一个 MongoDB CursorStream，我想将它传送到我编写的自定义 WritableStream（TypeScript）：

class NormalizerStream extends stream.Writable {
    constructor(opts?: stream.WritableOptions) {
        super(opts);
    }
    public _write(data: Buffer, encoding: string, callback: Function): void;
    public _write(data: string, encoding: string, callback: Function): void;
    public _write(data: any, encoding: string, callback: Function): void {
        console.log(data);
        callback();
    }
}

我打开一个到我的数据库的连接并从一个集合中检索所有文档，我想将它通过管道传输到我的 NormalizerStream 以规范化数据并将结果保存在另一个集合中。所以我正在这样做（JavaScript）：

Mongo.checkNative(function (db) {
    db.collection('users').find({}).stream().pipe(new NormalizerStream());
});

但是在运行代码的时候，总是出现这个错误：

TypeError: Invalid non-string/buffer chunk

JSON 对象是否有某种限制，我无法通过管道传输它们？还是我只是做错了什么？

【问题讨论】：

标签： node.js stream

【解决方案1】：

答案是，我显然需要实现转换流，而不是可写流。我仍然不太清楚为什么可写流不采用 JSON。

【讨论】：

【解决方案2】：

我在尝试解析来自 postgres 的 knex 数据时遇到了同样的问题。在查看节点源之后，事实证明您必须使用 objectMode: true 和 highWaterMark 构造您的 Writable Stream。然后您的 write 方法将获得 JSON 块。否则你会看到上面的错误。

   var writableStream = new stream.Writable(
  {
    objectMode : true,
    highWaterMark : 16,
    write : function(chunk, encoding, callback) {
      console.log("write", chunk);
      callback();
    },
    writev : function(chunk, callback) {
      console.log("writeev", chunk);
      callback();
    }
});

【讨论】：