使用streams2创建一个通过子进程传递数据的流

Question

假设我有一个函数BlackBox。api是这样的(实际上，|是管道)：

inputStream | BlackBox | outputStream

然而，BlackBox实际上是一个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