如何正确处理Node.js Transform流的end与close事件？

根因说明 #

这个问题和Transform流的实现逻辑无关，是Node.js流的内置运行机制导致的：

• Node.js可读流默认处于暂停模式，只有绑定data事件、调用resume()方法、或者通过pipe()对接下游消费者三种方式，才会切换为流动模式，开始消费内部缓存数据。
• 你第一段代码中，pipe()返回的Transform流没有对接任何下游消费者，流的内部缓存一直处于待消费状态，因此不会触发end、close这类生命周期结束的事件。
• 你添加data事件监听后相当于给Transform流增加了消费者，缓存排空后自然会触发所有结束事件。

正确实现方案 #

根据使用场景选择对应方案即可：

场景1：需要消费Transform流的输出 #

正常对接下游消费者即可，两种常用方式：

1. 继续用pipe()对接其他可写流：

const { Transform, Writable } = require('stream');
// 省略上游可读流定义
const transform = new Transform({
  objectMode: true,
  transform(item, encoding, callback) {
    console.debug(`transform ${item}`);
    this.push(item); // 必须调用push方法把处理后的数据推入输出缓存
    callback();
  }
})
readable2.pipe(transform).pipe(new Writable({
  objectMode: true,
  write(chunk, encoding, callback) {
    console.log(chunk);
    callback();
  }
}))

2. 手动绑定data事件监听消费数据，就是你第二段代码的写法。

场景2：Transform仅做中间逻辑处理，不需要输出数据 #

不需要绑定无用的data事件，直接调用resume()方法排空流缓存即可：

const readable2 = stream.Readable.from([4, 5, 6])
  .pipe(new stream.Transform({
    objectMode: true,
    transform(item, encoding, callback) {
      console.debug(`transform ${item}`);
      // 不需要调用this.push传递数据
      callback();
    }
  }))
  .on('end', () => console.log("transform is ended"))
  .on('close', () => console.log("transform is closed"))
  .resume(); // 主动触发流消费，排空缓存后自动触发结束事件

额外注意点 #

如果Transform流需要向下游传递数据，必须在transform方法中调用this.push(处理后的数据)，否则下游拿不到输出内容。

内容的提问来源于stack exchange，提问作者Valentyn Derkach