stream: make the pipeline callback mandatory

Right now when not adding a callback to the pipeline it could cause an uncaught exception if there is an error. Instead, just make the callback mandatory as mostly done in all other Node.js callback APIs so users explicitly have to decide what to do in such situations. PR-URL: #21054 Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: Anna Henningsen <[email protected]> Reviewed-By: Michaël Zasso <[email protected]> Reviewed-By: Trivikram Kamat <[email protected]> Reviewed-By: James M Snell <[email protected]>
nodejs · Jun 7, 2018 · 32c51f1 · 32c51f1
1 parent 505bfdc
commit 32c51f1
Show file tree

Hide file tree

Showing 3 changed files with 14 additions and 24 deletions.
diff --git a/doc/api/stream.md b/doc/api/stream.md
@@ -1340,14 +1340,14 @@ run().catch(console.error);
 rs.resume(); // drain the stream
 ```
 
-### stream.pipeline(...streams[, callback])
+### stream.pipeline(...streams, callback)
 <!-- YAML
 added: v10.0.0
 -->
 
 * `...streams` {Stream} Two or more streams to pipe between.
-* `callback` {Function} A callback function that takes an optional error
-  argument.
+* `callback` {Function} Called when the pipeline is fully done.
+  * `err` {Error}
 
 A module method to pipe between streams forwarding errors and properly cleaning
 up and provide a callback when the pipeline is complete.

diff --git a/lib/internal/streams/pipeline.js b/lib/internal/streams/pipeline.js
@@ -6,6 +6,7 @@
 let eos;
 
 const {
+  ERR_INVALID_CALLBACK,
   ERR_MISSING_ARGS,
   ERR_STREAM_DESTROYED
 } = require('internal/errors').codes;
@@ -19,11 +20,6 @@ function once(callback) {
   };
 }
 
-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';
 }
@@ -66,8 +62,11 @@ function pipe(from, to) {
 }
 
 function popCallback(streams) {
-  if (!streams.length) return noop;
-  if (typeof streams[streams.length - 1] !== 'function') return noop;
+  // Streams should never be an empty array. It should always contain at least
+  // a single stream. Therefore optimize for the average case instead of
+  // checking for length === 0 as well.
+  if (typeof streams[streams.length - 1] !== 'function')
+    throw new ERR_INVALID_CALLBACK();
   return streams.pop();
 }
 

diff --git a/test/parallel/test-stream-pipeline.js b/test/parallel/test-stream-pipeline.js
@@ -60,7 +60,7 @@ common.crashOnUnhandledRejection();
   }, /ERR_MISSING_ARGS/);
   assert.throws(() => {
     pipeline();
-  }, /ERR_MISSING_ARGS/);
+  }, /ERR_INVALID_CALLBACK/);
 }
 
 {
@@ -493,17 +493,8 @@ common.crashOnUnhandledRejection();
     }
   });
 
-  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');
+  assert.throws(
+    () => pipeline(read, transform, write),
+    { code: 'ERR_INVALID_CALLBACK' }
+  );
 }