more updates, nextTick the removable of listeners in tests

Author: phated

lib/sink.js

@@ -11,33 +11,43 @@ function hasListeners(stream) {
 }
 
 function sink(stream) {
+  var sinkAdded = false;
   var sinkStream = new Writable({
     objectMode: true,
     write: function(file, enc, cb) {
       cb();
     },
   });
 
+  function addSink() {
+    if (sinkAdded) {
+      return;
+    }
+
+    if (hasListeners(stream)) {
+      return;
+    }
+
+    sinkAdded = true;
+    stream.pipe(sinkStream);
+  }
+
   function removeSink(evt) {
     if (evt !== 'readable' && evt !== 'data') {
       return;
     }
 
     if (hasListeners(stream)) {
-      stream.unpipe(sinkStream);
+      sinkAdded = false;
+      return stream.unpipe(sinkStream);
     }
   }
 
   stream.on('newListener', removeSink);
   stream.on('removeListener', removeSink);
+  stream.on('removeListener', addSink);
 
-  return function() {
-    if (hasListeners(stream)) {
-      return;
-    }
-
-    stream.pipe(sinkStream);
-  };
+  return addSink;
 }
 
 module.exports = sink;

test/dest.js

@@ -1411,7 +1411,9 @@ describe('dest stream', function() {
 
     srcStream.pipe(countFiles).pipe(destStream);
 
-    destStream.removeListener('readable', noop);
+    process.nextTick(function() {
+      destStream.removeListener('readable', noop);
+    });
   });
 
   it('sinks the stream if all the data event handlers are removed', function(done) {
@@ -1444,7 +1446,9 @@ describe('dest stream', function() {
 
     srcStream.pipe(countFiles).pipe(destStream);
 
-    destStream.removeListener('data', onData);
+    process.nextTick(function() {
+      destStream.removeListener('data', onData);
+    });
   });
 
   it('should pass options to through2', function(done) {