node.js - 从nodejs中的可写流中暂停管道可读流的正确方法是什么？

Question

我正在编写一个模块，它是一个可写流。我想为我的用户实现管道接口(interface)。

如果发生错误，我需要暂停可读流并发出错误事件。然后，用户将决定 - 如果他没问题，他应该能够恢复数据处理。

var writeable = new BackPressureStream();
writeable.on('error', function(error){
    console.log(error);
    writeable.resume();
});

var readable = require('fs').createReadStream('somefile.txt');
readable.pipe.(writeable);

我看到 Node 为我们提供了 readable.pause() 方法，可以用来暂停可读流。但我无法从我的可写流模块中调用它:

var Writable = require('stream').Writable;

function BackPressureStream(options) {
    Writable.call(this, options);
}
require('util').inherits(BackPressureStream, Writable);

BackPressureStream.prototype._write = function(chunk, encoding, done) {
    done();
};

BackPressureStream.prototype.resume = function() {
    this.emit('drain');
}

如何在可写流中实现背压？

附:可以使用提供可读流作为参数的 pipe/unpipe 事件。但也有人说，对于管道流，暂停的唯一机会是将可读流从可写流中分离出来。

我做对了吗？在用户调用恢复之前，我必须解除可写流的管道？在用户调用 resume 之后，我应该通过管道返回可读流吗？

Answer 1

您所描述的内容已经由 pipe 方法本身实现。来自 Errors While Writing文档中的部分:

If a Readable stream pipes into a Writable stream when Writable emits an error, the Readable stream will be unpiped.

因此，作为可写流的实现者，您唯一的工作就是实现 _write 方法并在发生错误时发出错误。取消管道将由 Stream 模块自动处理。然后，如果模块的使用者认为错误是非关键的，则将可读流返回是他们的工作。他们可以这样做:

var writeable = new BackPressureStream();
var readable = require('fs').createReadStream('somefile.txt');

writeable.on('error', function(error) {
    // use pipe again, if error is not critical
    if (!error.critical) {
        readable.pipe(writeable);
    } else {
        readable.destroy(error);
    }
});

readable.pipe(writeable);

在你的模块内部:

BackPressureStream.prototype._write = function(chunk, encoding, done) {
    // call done with an error to emit 'error' event and unpipe readable stream
    done(new Error('BOOM'));
};