diff --git a/lib/internal/streams/pipeline.js b/lib/internal/streams/pipeline.js index edbfed3e516c05..493a464d07b295 100644 --- a/lib/internal/streams/pipeline.js +++ b/lib/internal/streams/pipeline.js @@ -25,8 +25,20 @@ let EE; let PassThrough; let createReadableStreamAsyncIterator; -function isRequest(stream) { - return stream && stream.setHeader && typeof stream.abort === 'function'; +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) { @@ -37,10 +49,22 @@ function destroyer(stream, reading, writing, final, callback) { eos(stream, { readable: reading, writable: writing }, (err) => { if (destroyed) return; destroyed = true; - 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 && reading && !writing && stream.writable) { + return callback(); + } + + if (err || !final || !stream.readable) { destroyImpl.destroyer(stream, err); } + callback(err); }); diff --git a/test/parallel/test-stream-pipeline.js b/test/parallel/test-stream-pipeline.js index 9939a16494499c..5175b7e07c53c2 100644 --- a/test/parallel/test-stream-pipeline.js +++ b/test/parallel/test-stream-pipeline.js @@ -995,3 +995,40 @@ 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()); + }); +} + +{ + // Might still want to be able to use the writable side + // of src. This is in the case where e.g. the Duplex input + // is not directly connected to its output. Such a case could + // happen when the Duplex is reading from a socket and then echos + // the data back on the same socket. + const src = new PassThrough(); + assert.strictEqual(src.writable, true); + const dst = new PassThrough(); + pipeline(src, dst, common.mustCall((err) => { + assert.strictEqual(src.writable, true); + assert.strictEqual(src.destroyed, false); + })); + src.push(null); +}