basic streams2 readable example

read_test.js

var util   = require("util");
var stream = require("stream");
var ee     = require("events").EventEmitter;

var Readable = stream.Readable;
var emitter  = new ee();

// basically just to show stuff going from one place to abother
var pushed = ['a', 'b', 'c', 'd', 'e'];
var pulled = [];
var endOfStream = false;

// Very base use of inherits
var MyReader = function(opts){
  Readable.call(this, opts);
};
util.inherits(MyReader, Readable);

MyReader.prototype._read = function (n, cb) {

  // When we're are the end of our pulls and pushes, end of stram
  if(endOfStream)
    return cb(null, null);

  // We wait for an event from our downstream and then pass it
  // down the line
  emitter.once('data_added', function() {
    // add a newline to show that we're doing stuff
    var thing = pulled.shift() + " -- \n";
    cb(null, thing);
  });
};

// Since we need a writable downstream, lets just use stdout
var readme = new MyReader();
readme.pipe(process.stdout);

// add stuff to be pulled every second and clear our interval and
// mark end of our stream
var meid = setInterval(function(){
  if(pushed.length == 0){
    endOfStream = true;
    clearInterval(meid);
  } else {
    pulled.push( pushed.shift() );
    emitter.emit('data_added'); // <--- incoming!
  }
}, 1000);