はじめてのReadableStream(stream2)

stream2のReadableStreamを書いてみました。

var stream = require('stream'),
    util = require('util');

function SpecialStream(options) {
  options || (options = {});
  options.objectMode = true;
  stream.Readable.call(this, options);

  this.count = 0;
}

util.inherits(SpecialStream, stream.Readable);

SpecialStream.prototype._read = function(size) {
  this.add_(this.count++ + ': ' + String(Math.random()) + '\n');

  if (this.count > 30000) {
    this.push('size: ' + String(size) + '\n');
    this.push(null);
  }
};

SpecialStream.prototype.add_ = function(val) {
  var that = this;

  !this.push(val) && setImmediate(function() {
    that.add_(val);
  });
};

var aaa = new SpecialStream();

aaa.pipe(process.stdout);

うーん、一応書いてみたけど……よくわからない。
stream1の時もそうだったけど、結局なにを実装すればいいのか全然わかんない。

pipeでは動くんだけど、EventEmitterスタイルでやると全然動かないんだこれが。
今までのdataイベントをそのままreadableイベントでemitしてあげるといいのかな？

追記：

--- index.js    2013-03-14 11:27:44.913590356 +0900
+++ index2.js   2013-03-14 11:28:00.673524659 +0900
@@ -30,4 +30,9 @@

 var aaa = new SpecialStream();

-aaa.pipe(process.stdout);
+aaa.on('readable', function() {
+  var buf;
+  while ((buf = aaa.read(1)) !== null) {
+    process.stdout.write(buf);
+  }
+});

こうですか先生わかりません（＞＜）

一応pipeしたときみたいには動いたけど……