-
-
Notifications
You must be signed in to change notification settings - Fork 31.9k
Streams: forward errors on pipe #1521
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -447,6 +447,7 @@ Readable.prototype._read = function(n) { | |
Readable.prototype.pipe = function(dest, pipeOpts) { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We should probably add this option to There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Probably I forgot about that one |
||
var src = this; | ||
var state = this._readableState; | ||
pipeOpts = pipeOpts || {}; | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. should we care that in a typical use-case this would create an arbitrary object? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. we also avoid an unnecessary conversion of undefined to a Boolean, I doubt it makes much difference |
||
|
||
switch (state.pipesCount) { | ||
case 0: | ||
|
@@ -462,7 +463,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) { | |
state.pipesCount += 1; | ||
debug('pipe count=%d opts=%j', state.pipesCount, pipeOpts); | ||
|
||
var doEnd = (!pipeOpts || pipeOpts.end !== false) && | ||
var doEnd = (pipeOpts.end !== false) && | ||
dest !== process.stdout && | ||
dest !== process.stderr; | ||
|
||
|
@@ -492,6 +493,12 @@ Readable.prototype.pipe = function(dest, pipeOpts) { | |
var ondrain = pipeOnDrain(src); | ||
dest.on('drain', ondrain); | ||
|
||
var forwardErrors = Boolean(pipeOpts.forwardErrors); | ||
|
||
if (forwardErrors) { | ||
src.on('error', onsrcerror); | ||
} | ||
|
||
function cleanup() { | ||
debug('cleanup'); | ||
// cleanup event handlers once the pipe is broken | ||
|
@@ -504,6 +511,10 @@ Readable.prototype.pipe = function(dest, pipeOpts) { | |
src.removeListener('end', cleanup); | ||
src.removeListener('data', ondata); | ||
|
||
if (forwardErrors) { | ||
src.removeListener('error', onsrcerror); | ||
} | ||
|
||
// if the reader is waiting for a drain event from this | ||
// specific writer, then it would cause it to never start | ||
// flowing again. | ||
|
@@ -526,6 +537,14 @@ Readable.prototype.pipe = function(dest, pipeOpts) { | |
} | ||
} | ||
|
||
// if src has an error and error forwarding is turned on | ||
// cause the error to be emited in dest and do suppress the | ||
// throwing behavior | ||
function onsrcerror(er) { | ||
debug('onsrcerror', er); | ||
dest.emit('error', er); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Do we care if this creates an infinite loop? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We should see how @domenic deals with this in whatwg streams There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. An object cannot be both readable and writable so the problem is avoided. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. But the writable half could forward to the readable half On Sat, Apr 25, 2015, 8:49 PM Domenic Denicola notifications@github.com
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. But that's fine. It means e.g. an echo server There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I could just change it from an on to a once There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. wait @chrisdickinson that won't be a problem because on an error the stream is unpiped so it will only ever pass one error through |
||
} | ||
|
||
// if the dest has an error, then stop piping into it. | ||
// however, don't suppress the throwing behavior for this. | ||
function onerror(er) { | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,58 @@ | ||
var common = require('../common'); | ||
var stream = require('stream'); | ||
var assert = require('assert'); | ||
|
||
(function errorsAreForwardes() { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Is this a typo? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. yes |
||
var writable = new stream.Writable({ | ||
write: function (chunk, _, cb) { | ||
process.nextTick(cb); | ||
} | ||
}); | ||
|
||
var called = 0; | ||
writable.on('error', function (){ | ||
called++; | ||
}); | ||
writable.on('unpipe', function () { | ||
process.nextTick(function () { | ||
assert.equal(called, 1); | ||
}); | ||
}); | ||
|
||
var readable = new stream.Readable({ | ||
read: function () { | ||
this.emit('error'); | ||
} | ||
}); | ||
readable.pipe(writable, { | ||
forwardErrors: true | ||
}) | ||
}()); | ||
|
||
|
||
(function CircularErrorsAreForwardes() { | ||
var passthrough1 = new stream.PassThrough(); | ||
var passthrough2 = new stream.PassThrough(); | ||
var called = 0; | ||
passthrough2.on('error', function (){ | ||
called++; | ||
}); | ||
passthrough2.on('unpipe', function () { | ||
process.nextTick(function () { | ||
assert.equal(called, 1); | ||
}); | ||
}); | ||
|
||
var readable = new stream.Readable({ | ||
read: function () { | ||
this.emit('error'); | ||
} | ||
}); | ||
readable.pipe(passthrough1, { | ||
forwardErrors: true | ||
}).pipe(passthrough2, { | ||
forwardErrors: true | ||
}).pipe(passthrough1, { | ||
forwardErrors: true | ||
}); | ||
}()) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Might reword this: