streams2 ReadStream that uses callbacks instead of a combination of 'readable' event listening and read()

readstream.js

var ReadableStream = require('stream').Readable;

var EMPTY_CALLBACK = function(n) {};

function ReadStream(cfg) {
  if (!(this instanceof ReadStream))
    return new ReadStream();

  var self = this;
  this._callbacks = [];
  this._rs = new ReadableStream();
  if (cfg && typeof cfg._read === 'function')
    this._rs._read = cfg._read;
  else
    this._rs._read = EMPTY_CALLBACK;
  this._rs.on('readable', function() {
    if (self._callbacks.length) {
      var cb, b = self._rs.read(self._callbacks[0][0]);
      if (b && self._callbacks[0][0] === b.length)
        (cb = self._callbacks.shift()[1]) && cb(b);
    }
  });
}
ReadStream.prototype.read = function(n, cb) {
  if (n > 0) {
    var b = this._rs.read(n);
    if (b === null)
      this._callbacks.push([n, cb]);
    else
      process.nextTick(function() { cb(b); });
  } else
    cb(null);
};
ReadStream.prototype.push = function(chunk) {
  var r = this._rs.push(chunk);
  this._rs.read(0);
  return r;
};
ReadStream.prototype.unshift = function(chunk) {
  var r = this._rs.unshift(chunk);
  this._rs.read(0);
  return r;
};
ReadStream.prototype.setEncoding = function(enc) {
  return this._rs.setEncoding(enc);
};

module.exports = ReadStream;

test.js

var ReadStream = require('./readstream');

var rs = new ReadStream();
rs.read(5, function(b) {
  console.log('5 bytes: ' + b.toString());
});
rs.read(6, function(b) {
  console.log('6 bytes: ' + b.toString());
});

var chunks = [
  new Buffer('he'),
  new Buffer('llo '),
  new Buffer('worl'),
  new Buffer('d')
], chn = 0;

while (chn < chunks.length && rs.push(chunks[chn++]));

// output:
// 5 bytes: 'hello'
// 6 bytes: ' world'