为什么Node.js Stream的pipeline回调不触发且Transform处理16个chunk后停止？

• 链路缺少最终Writable目的地
  pipeline要求流链路必须形成「Readable源头 → 中间多个Transform/Duplex双工流 → Writable目的地」的完整闭环。原代码仅传入Readable和Transform，最后一个节点是双工的Transform，其可读端没有任何消费者，直接引发两个连锁问题：
• 高水位线触发背压导致处理中断
  objectMode下所有流的默认highWaterMark为16个对象。当Transform的可写端缓冲攒满16个chunk后，会触发背压机制，停止向Readable请求后续数据；又因为可读端没人消费，缓冲永远无法排空，最终transform处理函数处理完16个chunk后就不再被调用。
• 链路未完成导致回调不执行
  pipeline的回调仅在全链路所有节点完成结束（收到end信号、所有缓冲数据排空）后才会触发。由于Transform的可读端无人消费，永远不会触发end事件，整个链路的结束流程无法完成，导致回调始终不执行。

只需在pipeline参数中新增一个objectMode的Writable作为最终目的地，负责消费Transform的可读端数据即可，修改后的完整可运行代码如下：

const { Readable, Transform, Writable, pipeline } = require('stream');

// 新增最终消费用的Writable流
const writable = new Writable({
    objectMode: true,
    write(chunk, encoding, callback) {
        // 无需额外处理数据时直接触发回调即可完成消费
        callback();
    }
});

const readable = new Readable({
    objectMode: true,
    read() {
        this.push(count++);
        if (count > 20) {
            this.push(null);
        }
    }
});
const transform = new Transform({
    objectMode: true,
    transform(chunk, encoding, callback) {
        data.push(chunk);
        console.log('transform - chunk: ', chunk.toString());
        callback(null, chunk);
    }
});

let count = 1, data = [];
pipeline(
    readable,
    transform,
    writable, // 新增Writable作为链路终点
    (error) => {
        if (error) console.log('pipeline callback - ERROR: ', error);
        else console.log('pipeline callback - data: ', data);
    }
);

可选优化：原Readable的read方法可优化为尽可能多推送数据直到达到高水位线，减少read方法的触发次数，提升性能，优化后的实现如下：

read() {
    let pushResult = true;
    // 循环推送直到push返回false（缓冲达到highWaterMark）或数据全部推完
    while (pushResult && count <= 20) {
        pushResult = this.push(count++);
    }
    // 所有数据推完后推送null发送结束信号
    if (count > 20) this.push(null);
}

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