如何实现基本节点 Stream.Readable 示例

Question

我正在尝试学习流，但在使其正常工作时遇到了一点问题。

对于此示例，我只想将静态对象推送到流中，并将其通过管道传输到我的服务器响应。

这是我到目前为止所拥有的，但很多都不起作用。如果我甚至可以将流输出到控制台，我就可以弄清楚如何将其管道传输到我的响应。

var Readable = require('stream').Readable;
var MyStream = function(options) {
  Readable.call(this);
};
MyStream.prototype._read = function(n) {
  this.push(chunk);
};
var stream = new MyStream({objectMode: true});
s.push({test: true});
request.reply(s);

Answer 1

您当前的代码存在一些问题。

1. 请求流很可能是缓冲区模式流：这意味着您无法将对象写入其中。幸运的是，您没有将选项传递给 Readable 构造函数，因此您的错误不会造成任何麻烦，但从语义上讲，这是错误的，不会产生预期的结果。
2. 调用 Readable 的构造函数，但不继承原型属性。您应该使用 util.inherits() 来子类Readable。
3. chunk变量未在代码示例中的任何位置定义。

这是一个工作示例：

var util = require('util');
var Readable = require('stream').Readable;
var MyStream = function(options) {
  Readable.call(this, options); // pass through the options to the Readable constructor
  this.counter = 1000;
};
util.inherits(MyStream, Readable); // inherit the prototype methods
MyStream.prototype._read = function(n) {
  this.push('foobar');
  if (this.counter-- === 0) { // stop the stream
    this.push(null);
  }
};
var mystream = new MyStream();
mystream.pipe(process.stdout);

Answer 2

以下是Paul Mougel使用ES6的答案的变体。它实现从正counter到0的倒计时流（计数器默认为 1000）。然后我们创建一个 100 的流counter，它被管道传输到可写流process.stdout：

const { Readable } = require('stream');
class CountdownStream extends Readable {
  constructor(counter, options) {
    super(options);
    this._counter = (!isNaN(counter) && counter > 0) ? counter : 1000;
  }
  _read(size) {
    if (this._counter == 0) {
      this.push(this._counter.toString());
      this.push(null);
    }
    else {
      this.push(this._counter + "n");
    }
    this._counter -= 1;
  }
}
const counter = new CountdownStream(100);
counter.pipe(process.stdout);