_stream_readable.diff

diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js
index e1636c8..e65efd7 100644
--- a/lib/_stream_readable.js
+++ b/lib/_stream_readable.js
@@ -544,7 +544,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
   // however, don't suppress the throwing behavior for this.
   function onerror(er) {
     debug('onerror', er);
-    unpipe();
+    unpipe(er);
     dest.removeListener('error', onerror);
     if (EE.listenerCount(dest, 'error') === 0)
       dest.emit('error', er);
@@ -573,9 +573,9 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
   }
   dest.once('finish', onfinish);
 
-  function unpipe() {
+  function unpipe(er) {
     debug('unpipe');
-    src.unpipe(dest);
+    src.unpipe(dest, er);
   }
 
   // tell the dest that it's being piped to
@@ -604,7 +604,13 @@ function pipeOnDrain(src) {
 }
 
 
-Readable.prototype.unpipe = function(dest) {
+Readable.prototype._emitUnpipe = function(dest, err) {
+  dest.emit('unpipe', this, err);
+  this.emit('unpipe', dest, err);
+};
+
+
+Readable.prototype.unpipe = function(dest, err) {
   var state = this._readableState;
 
   // if we're not piping anywhere, then do nothing.
@@ -625,7 +631,7 @@ Readable.prototype.unpipe = function(dest) {
     state.pipesCount = 0;
     state.flowing = false;
     if (dest)
-      dest.emit('unpipe', this);
+      this._emitUnpipe(dest, err);
     return this;
   }
 
@@ -640,7 +646,7 @@ Readable.prototype.unpipe = function(dest) {
     state.flowing = false;
 
     for (var i = 0; i < len; i++)
-      dests[i].emit('unpipe', this);
+      this._emitUnpipe(dests[i], err);
     return this;
   }
 
@@ -654,7 +660,7 @@ Readable.prototype.unpipe = function(dest) {
   if (state.pipesCount === 1)
     state.pipes = state.pipes[0];
 
-  dest.emit('unpipe', this);
+  this._emitUnpipe(dest, err);
 
   return this;
 };