创建一个流通过数据通过子流程与streams2

Question

假设我有一个功能​​。该API是这样的（其中|，事实上，管道）：

inputStream | BlackBox | outputStream 

然而，​​实际上是一个require('child_process').spawn的包装，所以实际上它看起来是这样的：

inputStream | BlackBox.Writable -> proc.stdin -> proc.stdout -> BlackBox.Readable | outputStream 

我可以用streams1轻松做到这一点，但我想了解streams2以及它如何更好。因此，我到目前为止的代码如下：

var Duplex = require('stream').Duplex 
var spawn = require('child_process').spawn 
var util = require('util') 

util.inherits(BlackBox, Duplex) 

function BlackBox() { 
    Duplex.call(this) 

    // Example process 
    this.proc = spawn('convert', ['-', ':-']) 

    var that = this 
    this.proc.stdout.on('end', function() { 
    that.push(null) 
    }) 
} 

BlackBox.prototype._write = function (chunk, encoding, callback) { 
    return this.proc.stdin.write(chunk, encoding, callback) 
} 

BlackBox.prototype.end = function (chunk, encoding, callback) { 
    return this.proc.stdin.end(chunk, encoding, callback) 
} 

BlackBox.prototype._read = function (size) { 
    var that = this 

    this.proc.stdout.on('readable', function() { 
    var chunk = this.read(size) 
    if (chunk === null) 
     that.push('') 
    else 
     that.push(chunk) 
    }) 
} 

我在这里做了什么不对吗？

我最关心的是从文档以下摘录于readable._read(size)：

数据可用时，将其放入读队列调用readable.push（块）。如果push返回false，那么你应该停止阅读。当再次调用_read时，您应该开始推送更多数据。

我该如何“停止阅读”？

为了清楚起见，我想要回压和节流来处理。

Answer 1

艾萨克斯基本上做了一个例子：https://github.com/isaacs/duplex-passthrough