diff.patch

diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js
index a9d9fc6..16668ee 100644
--- a/lib/_stream_readable.js
+++ b/lib/_stream_readable.js
@@ -70,6 +70,11 @@ function ReadableState(options, stream) {
   this.emittedReadable = false;
   this.readableListening = false;
 
+  // Track whether the stream should close when dest closes:
+  // this is a fuzzy check to see if it looks like the user
+  // is handling it themselves.
+  this.shouldCloseOnDestClose = !!stream.close;
+
   // Crypto is kind of old and crusty.  Historically, its default string
   // encoding is 'binary' so we have to make this configurable.
   // Everything else in the universe uses 'utf8', though.
@@ -471,6 +476,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
       state.pipes = dest;
       break;
     case 1:
+      state.shouldCloseOnDestClose = false;
       state.pipes = [state.pipes, dest];
       break;
     default:
@@ -568,6 +574,10 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
   function onclose() {
     dest.removeListener('finish', onfinish);
     unpipe();
+
+    if(state.shouldCloseOnDestClose) {
+      src.close()
+    }
   }
   dest.once('close', onclose);
   function onfinish() {
@@ -692,6 +702,10 @@ Readable.prototype.on = function(ev, fn) {
     }
   }
 
+  if(ev === 'unpipe') {
+    this.shouldCloseOnDestClose = false;
+  }
+
   return res;
 };
 Readable.prototype.addListener = Readable.prototype.on;