Stream Sample for blog

01.mystream.js

var stream = require('stream')
  , util = require('util')
  , log = console.log.bind(console)
  ;

// 本来は 'drain','error','close','pipe' イベントが必要
function MyStream() {
  this.writable = true;
  this.buf = [];
}

// 継承、詳細は util.inherits を参照
util.inherits(MyStream, stream.Stream);

MyStream.prototype.write = function(data) {
  var data = data.toString().trim();
  log('write:', data);
  this.buf.push(data);
  return true;
};

MyStream.prototype.end = function(data) {
  log('end:', data);
  if (data) this.write(data);
  this.writable = false;
  log('\nresult:', this.buf.join(''));
};

MyStream.prototype.destroy = function() {};
MyStream.prototype.destroySoon = function() {};

module.exports = MyStream;

if (require.main === module) {
  var mystream = new MyStream();

  // 標準入力をパイプする
  process.stdin.pipe(mystream);
  // 読み込み開始
  process.stdin.resume();
}

02.timerstream.js

var stream = require('stream')
  , util = require('util')
  , log = console.log.bind(console)
  ;

// 本来は 'data', 'end', 'error', 'close' イベントが必要
function TimerStream() {
  this.readable = true;
  this.t = 0;
  this.timer = null;
  this.piped = false;
}

// 継承、詳細は util.inherits を参照
util.inherits(TimerStream, stream.Stream);

TimerStream.prototype.resume = function() {
  this.timer = setInterval(function() {
    this.t++;
    if (this.t > 4) {
      return this.emit('end');
    }
    this.emit('data', this.t.toString());
  }.bind(this), 1000);
};

TimerStream.prototype.pause = function() {
  clearInterval(this.timer);
};

TimerStream.prototype.pipe = function(dest) {
  this.piped = true;

  // ここでは stream.Stream.prototype.pipe.apply(this, arguments); もok
  this.on('data', function(data) {
    dest.write(data);
  });
};

TimerStream.prototype.setEncoding = function(encoding) {};
TimerStream.prototype.destroy = function() {};
TimerStream.prototype.destroySoon = function() {};

module.exports = TimerStream;

if (require.main === module) {
  var timerStream = new TimerStream();
  timerStream.pipe(process.stdout);
  timerStream.resume();
}

03.pipesample.js

var TimerStream = require('./timerstream')
  , MyStream = require('./mystream')
  , ts = new TimerStream()
  , ms = new MyStream()
  ;

// パイプで繋ぐ
ts.pipe(ms);
// 読み込みを開始
ts.resume();