stream: ensure Stream.pipeline re-throws errors without callback

phated · addaleax · commit d1e81b0f17a3 · 2018-05-31T11:34:50.000+02:00
Fixes an issue where Stream.pipeline wouldn't re-throw errors on a stream if no callback was specified, thus swallowing said errors. Fixes: #20303 PR-URL: #20437 Reviewed-By: Anna Henningsen <anna@addaleax.net> Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
diff --git a/lib/internal/streams/pipeline.js b/lib/internal/streams/pipeline.js
@@ -19,7 +19,10 @@ function once(callback) {
 };
 }
 
-function noop() {}
+function noop(err) {
+ // Rethrow the error if it exists to avoid swallowing it
+ if (err) throw err;
+}
 
 function isRequest(stream) {
 return stream.setHeader && typeof stream.abort === 'function';
diff --git a/test/parallel/test-stream-pipeline.js b/test/parallel/test-stream-pipeline.js
@@ -142,7 +142,7 @@ common.crashOnUnhandledRejection();
 }
 });
 
- pipeline(rs, res);
+ pipeline(rs, res, () => {});
 });
 
 server.listen(0, () => {
@@ -177,7 +177,7 @@ common.crashOnUnhandledRejection();
 })
 });
 
- pipeline(rs, res);
+ pipeline(rs, res, () => {});
 });
 
 server.listen(0, () => {
@@ -206,7 +206,7 @@ common.crashOnUnhandledRejection();
 })
 });
 
- pipeline(rs, res);
+ pipeline(rs, res, () => {});
 });
 
 let cnt = 10;
@@ -481,3 +481,35 @@ common.crashOnUnhandledRejection();
 
 run();
 }
+
+{
+ const read = new Readable({
+ read() {}
+ });
+
+ const transform = new Transform({
+ transform(data, enc, cb) {
+ cb(new Error('kaboom'));
+ }
+ });
+
+ const write = new Writable({
+ write(data, enc, cb) {
+ cb();
+ }
+ });
+
+ read.on('close', common.mustCall());
+ transform.on('close', common.mustCall());
+ write.on('close', common.mustCall());
+
+ process.on('uncaughtException', common.mustCall((err) => {
+ assert.deepStrictEqual(err, new Error('kaboom'));
+ }));
+
+ const dst = pipeline(read, transform, write);
+
+ assert.strictEqual(dst, write);
+
+ read.push('hello');
+}

-Original file line number
+Diff line change
 };
+}
 -functionnoop(){}
 +functionnoop(err){
 +// Rethrow the error if it exists to avoid swallowing it
 +if(err)throwerr;
 +}
 functionisRequest(stream){
 returnstream.setHeader&&typeofstream.abort==='function';
-Original file line number
+Diff line change
+}
 });
 -pipeline(rs,res);
 +pipeline(rs,res,()=>{});
 });
 server.listen(0,()=>{
 })
 });
 -pipeline(rs,res);
 +pipeline(rs,res,()=>{});
 });
 server.listen(0,()=>{
 })
 });
 -pipeline(rs,res);
 +pipeline(rs,res,()=>{});
 });
 letcnt=10;
 run();
+}
++
 +{
 +constread=newReadable({
 +read(){}
 +});
++
 +consttransform=newTransform({
 +transform(data,enc,cb){
 +cb(newError('kaboom'));
 +}
 +});
++
 +constwrite=newWritable({
 +write(data,enc,cb){
 +cb();
 +}
 +});
++
 +read.on('close',common.mustCall());
 +transform.on('close',common.mustCall());
 +write.on('close',common.mustCall());
++
 +process.on('uncaughtException',common.mustCall((err)=>{
 +assert.deepStrictEqual(err,newError('kaboom'));
 +}));
++
 +constdst=pipeline(read,transform,write);
++
 +assert.strictEqual(dst,write);
++
 +read.push('hello');
 +}