stream through samples

buffering.js

var through = require('through');
var log = console.log.bind(console);

var ts = through(function (data) { // data pre event
  this.queue(data);
}, function () { // end pre event
  this.queue(null);
});

var data = [];
ts.on('data', function (buf) {
  log('ondata', buf);
  data.push(buf);
});
ts.on('end', function () {
  log('onend');
});

ts.write(1);
ts.write(2);
ts.write(3);
log('write 1, 2, 3 : data = ', data);

ts.pause();
ts.write(4);
ts.write(5);
ts.write(6);
log('pause, write 4, 5, 6 : data = ', data);

ts.resume();
log('resume : data = ', data);

ts.pause();
log('pause : data = ', data);

ts.end();
log('end : ', data);

ts.resume();
log('resume : data = ', data);

end.js

var through = require('through');
var log = console.log.bind(console);

var ts = through();

ts.on('end', function () {
  log('onend');
})

ts.on('close', function () {
  log('onclose');
})

ts.write(1);
ts.write(2);
ts.write(3);
ts.end();

readable.js

var through = require('through');

var ts = through();
var times = 0;
var iv = setInterval(function () {
  ts.write(times + '\n');
  if (++times === 5) {
    ts.end();
    clearInterval(iv);
  }
}, 1000);

ts.pipe(process.stdout);

writable.js

var fs = require('fs');
var through = require('through');

var bytes = 0;
fs.createReadStream('/etc/passwd').pipe(through(function (buf) {
  bytes += buf.length;
  this.queue(buf);
}, function () {
  console.log(bytes);
  this.queue(null);
}))