gistfile1.js

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

var readable = Readable();
readable._read = function(){
  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);

setTimeout(function(){
  writable.end();
}, 1000);

output

1391476923961
1391476924067
1391476924168
1391476924269
1391476924371
1391476924472
1391476924573
1391476924673
1391476924774

[02:38:53] rvagg> juliangruber: what's the use-case here btw?
[02:47:02] rvagg> juliangruber: https://gist.github.com/rvagg/8796067
[02:47:14] rvagg> juliangruber: normally the signal doesn't go back up from writable to the piped readable
[02:47:36] rvagg> writable.on('finish') -> readable.push(null) and clear the in-process timer so it doesn't bork
[02:49:39] rvagg> juliangruber: readable should be listening to 'finish' itself and cleaning up, I think the fact that read() keeps on returning available stuff on read(n) means sends the signal that it's not finished, hence the need to finish it manually
[02:50:11] rvagg> juliangruber: but you can imagine a use-case here, a readable that keeps on reading, pipe it to an endpoint until that endpoint has had enough, unpipe and pipe it to another endpoint and the buffer stays intact