harden event handling

Author: phated

lib/sink.js

@@ -2,6 +2,10 @@
 
 var Writable = require('readable-stream/writable');
 
+function listenerCount(stream, evt) {
+  return stream.listeners(evt).length;
+}
+
 function sink(stream) {
   var sinkStream = new Writable({
     objectMode: true,
@@ -10,9 +14,26 @@ function sink(stream) {
     },
   });
 
+  stream.on('removeListener', function() {
+    if (listenerCount(stream, 'readable') || listenerCount(stream, 'data')) {
+      stream.unpipe(sinkStream);
+    }
+  });
+
+  stream.on('unpipe', function() {
+    if (!(listenerCount(stream, 'readable') || listenerCount(stream, 'data'))) {
+      stream.pipe(sinkStream);
+    }
+  });
+
+  stream.on('newListener', function() {
+    if (listenerCount(stream, 'readable') || listenerCount(stream, 'data')) {
+      stream.unpipe(sinkStream);
+    }
+  });
+
   return function() {
-    // Respect readable listeners on the underlying stream
-    if (stream.listeners('readable').length > 0) {
+    if (listenerCount(stream, 'readable') || listenerCount(stream, 'data')) {
       return;
     }