mirror of
https://github.com/zebrajr/node.git
synced 2026-01-15 12:15:26 +00:00
stream: ensure pipeline always destroys streams
There was an edge case where an incorrect assumption was made in regardos whether eos/finished means that the stream is actually destroyed or not. PR-URL: https://github.com/nodejs/node/pull/31940 Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: Ruben Bridgewater <ruben@bridgewater.de> Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
This commit is contained in:
@@ -27,27 +27,20 @@ let createReadableStreamAsyncIterator;
|
||||
|
||||
function destroyer(stream, reading, writing, callback) {
|
||||
callback = once(callback);
|
||||
|
||||
let closed = false;
|
||||
stream.on('close', () => {
|
||||
closed = true;
|
||||
});
|
||||
let destroyed = false;
|
||||
|
||||
if (eos === undefined) eos = require('internal/streams/end-of-stream');
|
||||
eos(stream, { readable: reading, writable: writing }, (err) => {
|
||||
if (err) return callback(err);
|
||||
closed = true;
|
||||
callback();
|
||||
});
|
||||
|
||||
let destroyed = false;
|
||||
return (err) => {
|
||||
if (closed) return;
|
||||
if (destroyed) return;
|
||||
destroyed = true;
|
||||
|
||||
destroyImpl.destroyer(stream, err);
|
||||
callback(err);
|
||||
});
|
||||
|
||||
return (err) => {
|
||||
if (destroyed) return;
|
||||
destroyed = true;
|
||||
destroyImpl.destroyer(stream, err);
|
||||
callback(err || new ERR_STREAM_DESTROYED('pipe'));
|
||||
};
|
||||
}
|
||||
|
||||
@@ -763,7 +763,10 @@ const { promisify } = require('util');
|
||||
s.emit('data', 'asd');
|
||||
s.emit('end');
|
||||
});
|
||||
s.close = common.mustCall();
|
||||
// 'destroyer' can be called multiple times,
|
||||
// once from stream wrapper and
|
||||
// once from iterator wrapper.
|
||||
s.close = common.mustCallAtLeast(1);
|
||||
let ret = '';
|
||||
pipeline(s, async function(source) {
|
||||
for await (const chunk of source) {
|
||||
@@ -909,3 +912,13 @@ const { promisify } = require('util');
|
||||
assert.strictEqual(err.message, 'kaboom');
|
||||
}));
|
||||
}
|
||||
|
||||
{
|
||||
const src = new PassThrough({ autoDestroy: false });
|
||||
const dst = new PassThrough({ autoDestroy: false });
|
||||
pipeline(src, dst, common.mustCall(() => {
|
||||
assert.strictEqual(src.destroyed, true);
|
||||
assert.strictEqual(dst.destroyed, true);
|
||||
}));
|
||||
src.end();
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user