exp.txt

native-stream.js uses the reference implementation, compiled with babel to run
on latest io.js.

stream-bluebird.js is exactly the same as native-stream.js except it uses the
latest bluebird Promise implementation.

node-stream.js uses the node stream implementation and forced to be async by
using process.nextTick() (which is functionally equivalent to the microtask
processor)

node-stream.js:     ~480ns per call
native-stream.js:   ~7.8us per call  15.25x slower
bluebird-stream.js: ~2.5us per call    4.2x slower


also have transform streams being piped.

node-transform.js:     ~1800ns per call
native-transform.js:   ~53.1us per call 28.5x slower
transform-bluebird.js: ~14.7us per call 7.2x slower

native-stream.js

'use strict';

// Uncomment this for bluebird-stream.js
//Promise = require('bluebird');

let _interopRequire = function (obj) { return obj && obj.__esModule ? obj["default"] : obj; };
let ReadableStream = _interopRequire(require("../lib/readable-stream"));

const ITER = 1e5;
let i = ITER;
let b = new Buffer(1024 * 10);
let t = process.hrtime();

let rs = new ReadableStream({
  start: function start(enqueue, close, error) {
    enqueue(b);
  },
  pull: function pull(enqueue, close) {
    if (--i > 0)
      enqueue(b);
    else if (i-- === 0)
      printTime(process.hrtime(t));
  }
});

rs.ready.then(function read(v) {
  if (rs.state === 'readable')
    rs.read();
  rs.ready.then(read);
});


function printTime(t) {
  console.log(((t[0] * 1e9 + t[1]) / ITER).toFixed(1) + ' ns/op');
}

native-transform.js

'use strict';

// Uncomment this for bluebird-transform.js
//Promise = require('bluebird').Promise;

let _interopRequire = function (obj) { return obj && obj.__esModule ? obj["default"] : obj; };
let ReadableStream = _interopRequire(require("../lib/readable-stream"));
let TransformStream = _interopRequire(require("../lib/transform-stream"));
let WritableStream = _interopRequire(require("../lib/writable-stream"));

const ITER = 1e5;
let i = ITER;
let b = new Buffer(1024 * 16);
let t = process.hrtime();

let ts = new TransformStream({
  transform: function transform(chunk, enqueue, done) {
    enqueue(chunk);
    done();
  }
});

let ws = new WritableStream();

let rs = new ReadableStream({
  start: function start(enqueue, close, error) {
    enqueue(b);
  },
  pull: function pull(enqueue, close) {
    if (--i > 0) {
      enqueue(b);
    } else if (i-- === 0) {
      printTime(process.hrtime(t));
    }
  }
});


rs.pipeThrough(ts).pipeTo(ws);


function printTime(t) {
  console.log(((t[0] * 1e9 + t[1]) / ITER).toFixed(1) + ' ns/op');
}

node-stream.js

'use strict';

let Readable = require('stream').Readable;
let util = require('util');

const ITER = 1e5;
let i = ITER;
let b = new Buffer(1024 * 10);
let t = process.hrtime();

function RS() {
  Readable.call(this);
  this.push(b);
}
util.inherits(RS, Readable);

RS.prototype._read = function _read(size) {
  if (--i > 0)
    this.push(b);
  else if (i-- === 0)
    printTime(process.hrtime(t));
};


let rs = new RS();

rs.on('readable', function() {
  let self = this;
  process.nextTick(function() {
    while (self.read());
  });
});


function printTime(t) {
  console.log(((t[0] * 1e9 + t[1]) / ITER).toFixed(1) + ' ns/op');
}

node-transform.js

'use strict';

let Readable = require('stream').Readable;
let Transform = require('stream').Transform;
let Writable = require('stream').Writable;
let util = require('util');

const ITER = 1e5;
let i = ITER;
let b = new Buffer(1024 * 16);
let t = process.hrtime();

function RS() {
  Readable.call(this);
  this.push(b);
}
util.inherits(RS, Readable);

RS.prototype._read = function _read(size) {
  if (--i > 0)
    this.push(b);
};


function TS() {
  Transform.call(this);
}
util.inherits(TS, Transform);

TS.prototype._transform = function _transform(data, encoding, callback) {
  this.push(data);
  callback();
};


function WS() {
  Writable.call(this);
}
util.inherits(WS, Writable);

WS.prototype._write = function _write(chunk, encoding, callback) {
  if (i > 0)
    callback();
  else if (i-- === 0)
    printTime(process.hrtime(t));
};


let rs = new RS();
let ts = new TS();
let ws = new WS();

rs.pipe(ts).pipe(ws);


function printTime(t) {
  console.log(((t[0] * 1e9 + t[1]) / ITER).toFixed(1) + ' ns/op');
}