diff --git a/lib/internal/streams/pipeline.js b/lib/internal/streams/pipeline.js index 9f36d879ea6589..a446d5792e64bc 100644 --- a/lib/internal/streams/pipeline.js +++ b/lib/internal/streams/pipeline.js @@ -25,10 +25,32 @@ let EE; let PassThrough; let createReadableStreamAsyncIterator; +function isIncoming(stream) { + return ( + stream.socket && + typeof stream.complete === 'boolean' && + ArrayIsArray(stream.rawTrailers) && + ArrayIsArray(stream.rawHeaders) + ); +} + +function isOutgoing(stream) { + return ( + stream.socket && + typeof stream.setHeader === 'function' + ); +} + function destroyer(stream, reading, writing, final, callback) { const _destroy = once((err) => { - const readable = stream.readable || isRequest(stream); - if (err || !final || !readable) { + if (!err && (isIncoming(stream) || isOutgoing(stream))) { + // http/1 request objects have a coupling to their response and should + // not be prematurely destroyed. Assume they will handle their own + // lifecycle. + return callback(); + } + + if (err || !final || !stream.readable) { destroyImpl.destroyer(stream, err); } callback(err); @@ -71,10 +93,6 @@ function popCallback(streams) { return streams.pop(); } -function isRequest(stream) { - return stream.setHeader && typeof stream.abort === 'function'; -} - function isPromise(obj) { return !!(obj && typeof obj.then === 'function'); } diff --git a/test/parallel/test-stream-pipeline.js b/test/parallel/test-stream-pipeline.js index efb363f83b743d..6629e749238bda 100644 --- a/test/parallel/test-stream-pipeline.js +++ b/test/parallel/test-stream-pipeline.js @@ -1011,3 +1011,24 @@ const { promisify } = require('util'); assert.strictEqual(res, ''); })); } + +{ + const server = http.createServer((req, res) => { + req.socket.on('error', common.mustNotCall()); + pipeline(req, new PassThrough(), (err) => { + assert.ifError(err); + res.end(); + server.close(); + }); + }); + + server.listen(0, () => { + const req = http.request({ + method: 'PUT', + port: server.address().port + }); + req.end('asd123'); + req.on('response', common.mustCall()); + req.on('error', common.mustNotCall()); + }); +}