streams2: Unpipe on dest.emit('close')

lib/_stream_readable.js

@@ -350,11 +350,22 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
   // if the dest has an error, then stop piping into it.
   // however, don't suppress the throwing behavior for this.
   dest.once('error', function(er) {
-    src.unpipe(dest);
+    unpipe();
     if (dest.listeners('error').length === 0)
       dest.emit('error', er);
   });
 
+  // if the dest emits close, then presumably there's no point writing
+  // to it any more.
+  dest.on('close', unpipe);
+  dest.on('finish', function() {
+    dest.removeListener('close', unpipe);
+  });
+
+  function unpipe() {
+    src.unpipe(dest);
+  }
+
   // tell the dest that it's being piped to
   dest.emit('pipe', src);