cat.js

function cat(inputs) {
  var current, cb;
  var output = { read: read };
  return output;
  
  function read(callback) {
    if (!current) shift();
    cb = callback;
    current.read(onRead);
  }

  function onRead(data) {
    var callback = db;
    cb = null;
    return callback(data);
  }

  function shift() {
    current = inputs.shift();
    if (!current) return output.onEnd();
    current.onEnd = shift;
    current.onError = output.onError
    if (cb) current.read(onRead);
  }
}
  
  

filter.js

function transform(input) {
  return {
    read: read,
    set onError(fn) { input.onError = fn; },
    set onEnd(fn) { input.onEnd = fn; }
  };
  function read(callback) {
    input.read(function (data) {
      callback(data.toUpperCase());
    });
  }
}

/////////////////////////////////////////////////////

// Consuming
var parent = someStream();
var child = transform(parent);
child.onEnd = onEnd;
child.onError = onError;
return child.read(onRead);
function onRead(data) {
}
function onEnd() {
}
function onError(err) {
}

merge.js

function merge(inputs) {
  // If any input emits an error, this flag gets set to true.
  var dead = false;
  // The queue of data ready to be read.
  var queue = [];
  // The output stream
  var output = { read: read };

  inputs.forEach(prepInput);
  return output;
  
  
  function prepInput(input) {
    var done = false;
    input.onError = onError;
    input.onEnd = onEnd;
    return input.read(onRead);
    
  }

  function onError(err) {
    output.onError(err);
  }

new-stream.js

var stream = makeMagicStream();

stream.onEnd = onEnd;
stream.onError = onError;
stream.onRead = onRead;
return stream.read(onRead);

function onRead(data) {
  console.log(data);
  return stream.read(onRead);
}

function onError(err) {
  throw err;
}

function onEnd() {
  console.log("Done");
}