将数据放回可读流

Question

TL;DR 如何从流中读取一些数据，然后将其放回原处，以便其他使用者获得相同的data事件？

下面是一个可读的流，它流式传输 1...无限：

var Readable = require('stream').Readable;
var readable = new Readable();
var c = 0;
readable._read = function () {
    var self = this;
    setTimeout(function () {
        self.push((++c).toString());
    }, 500);
};

我想读取第一个data事件，查看数据，然后将流"重置"到其原始状态，并允许其他另一个data侦听器使用第一个事件，就好像它从未发生过一样。我认为unshift()是正确的方法，正如文档中所说：

可读.unshift（块）#

块缓冲区 |要取消移动到读取队列的数据块字符串 这在流被 解析器，需要"取消消费"它拥有的一些数据 乐观地从源中拉出，以便流可以 传给其他方。

这听起来很适合我的需求，但它并没有按照我的期望工作：

...
readable.once('data', function (d) {
    console.log(d.toString());              // Outputs 1
    readable.unshift(d);                    // Put the 1 back on the stream
    readable.on('data', function (d) {
        console.log(d.toString());          // Heh?! Outputs 2, how about 1?
    });
});

Answer 1

所以我找到了答案：

调用 stream.unshift() if 时，如果流处于流动模式，则会立即发出数据事件。因此，当我在示例中添加侦听器时，船已经航行了。

readable.unshift(d);                  // emits 'data' event
readable.on('data', function (d) {    // missed `data` event
    console.log(d.toString());
});

有几种方法可以使其按我的预期工作：

1） 在取消移位之前添加新侦听器：

readable.once('data', function (d) {
    console.log(d.toString());              // Outputs 1
    readable.on('data', function (d) {
        console.log(d.toString());          // Outputs 1,1,2,3...
    });
    readable.unshift(d);                    // Put the 1 back on the stream
});

2） 暂停和恢复流：

readable.once('data', function (d) {
    console.log(d.toString());              // Outputs 1
    readable.pause();                       // Stops the stream from flowing
    readable.unshift(d);                    // Put the 1 back on the stream
    readable.on('data', function (d) {
        console.log(d.toString());          // Outputs 1,1,2,3...
    });
    readable.resume();                      // Start the stream flowing again
});