如何创建unshift功能正常的Node.js Duplex流？求排查代码问题

你的代码核心问题在于直接替换了Duplex实例的read和on方法为外部可读流的方法，导致unshift存入的数据无法被读取：

• duplex.unshift()会把数据存入Duplex自身的内部缓冲区，但你替换后的duplex.read()实际调用的是外部readable的read方法，该方法只会读取外部流的缓冲区，完全不处理Duplex自己的缓冲区，所以返回null。
• 替换duplex.on还会破坏Duplex自身的事件机制，导致诸如readable这类关键事件的监听逻辑混乱。

修复后的代码 #

it("unshift duplex", async function () {
    function splice(readable: NodeJS.ReadableStream, writable: NodeJS.WritableStream): Duplex {
        const duplex = new Duplex({
            write(chunk, encoding, callback) {
                writable.write(chunk, encoding, callback);
            },

            final(callback) {
                writable.end(callback);
            },

            // 实现Duplex的_read方法，负责从外部可读流拉取数据到自身缓冲区
            _read(size) {
                const chunk = readable.read(size);
                if (chunk !== null) {
                    this.push(chunk);
                } else {
                    // 外部流暂时无数据时，监听readable事件再尝试拉取
                    readable.once('readable', () => {
                        this._read(size);
                    });
                }
            }
        });

        // 转发外部流的end事件，结束Duplex的可读端
        readable.on('end', () => {
            duplex.push(null);
        });

        // 转发错误事件到Duplex
        readable.on('error', err => duplex.emit('error', err));
        writable.on('error', err => duplex.emit('error', err));

        return duplex;
    }

    readable = new PassThrough();
    writable = new PassThrough();
    duplex = splice(readable, writable);

    duplex.unshift(Buffer.from([1, 2, 3]))
    const result = duplex.read()
    expect(result).toEqual(Buffer.from([1, 2, 3])) // 测试通过
})

修复说明 #

1. 实现_read方法：这是Duplex可读端的核心逻辑，负责从外部可读流拉取数据并推入Duplex自身的缓冲区，确保read操作能优先读取unshift存入的数据，再去拉外部流的内容。
2. 转发关键事件：将外部流的end、error事件转发到Duplex，保证流的生命周期和错误处理符合预期。
3. 不替换原生方法：保留Duplex自身的read和on方法，维持其内部缓冲区和事件机制的正常运转。

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