Skip to content

Commit

Permalink
stream: make the pipeline callback mandatory
Browse files Browse the repository at this point in the history
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 <matteo.collina@gmail.com>
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Michaël Zasso <targos@protonmail.com>
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
  • Loading branch information
BridgeAR authored and addaleax committed Jun 7, 2018
1 parent 505bfdc commit 32c51f1
Show file tree
Hide file tree
Showing 3 changed files with 14 additions and 24 deletions.
6 changes: 3 additions & 3 deletions doc/api/stream.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
13 changes: 6 additions & 7 deletions lib/internal/streams/pipeline.js
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
let eos;

const {
ERR_INVALID_CALLBACK,
ERR_MISSING_ARGS,
ERR_STREAM_DESTROYED
} = require('internal/errors').codes;
Expand All @@ -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';
}
Expand Down Expand Up @@ -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();
}

Expand Down
19 changes: 5 additions & 14 deletions test/parallel/test-stream-pipeline.js
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,7 @@ common.crashOnUnhandledRejection();
}, /ERR_MISSING_ARGS/);
assert.throws(() => {
pipeline();
}, /ERR_MISSING_ARGS/);
}, /ERR_INVALID_CALLBACK/);
}

{
Expand Down Expand Up @@ -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' }
);
}

0 comments on commit 32c51f1

Please sign in to comment.