gistfile1.js

var Readable = require('stream').Readable;
var Writable = require('stream').Writable;
 
var readable = Readable();

readable._read = function(){
  readable._t = setTimeout(function(){
    readable.push(''+Date.now());
  }, 100);
}

readable.on('end', function(){ console.log('end') })
readable.on('finish', function(){ console.log('finish') })
readable.on('close', function(){ console.log('close') })
 
var writable = Writable();
  writable._write = function(chunk, _, done){
  console.log(chunk.toString());
  done();
};
 
readable.pipe(writable)
writable.on('finish', function () {
  clearTimeout(readable._t)
  readable.push(null)
})
 
setTimeout(function(){
  writable.end();
}, 1000);

output

1391476923961
1391476924067
1391476924168
1391476924269
1391476924371
1391476924472
1391476924573
1391476924673
1391476924774