diff --git a/index.js b/index.js index b1a19067..3a0c18e3 100644 --- a/index.js +++ b/index.js @@ -14,10 +14,12 @@ * @private */ +var { finished } = require('node:stream') var Negotiator = require('negotiator') var bytes = require('bytes') var compressible = require('compressible') var debug = require('debug')('compression') +const isFinished = require('on-finished').isFinished var onHeaders = require('on-headers') var vary = require('vary') var zlib = require('zlib') @@ -215,7 +217,12 @@ function compression (options) { // compression stream.on('data', function onStreamData (chunk) { + if (isFinished(res)) { + debug('response finished') + return + } if (_write.call(res, chunk) === false) { + debug('pausing compression stream') stream.pause() } }) @@ -227,6 +234,15 @@ function compression (options) { _on.call(res, 'drain', function onResponseDrain () { stream.resume() }) + + // In case the stream is paused when the response finishes (e.g. because + // the client cuts the connection), its `drain` event may not get emitted. + // The following handler is here to ensure that the stream gets resumed so + // it ends up emitting its `end` event and calling the original + // `res.end()`. + finished(res, function onResponseFinished () { + stream.resume() + }) }) next() diff --git a/package.json b/package.json index 707c4aa7..449cbe23 100644 --- a/package.json +++ b/package.json @@ -14,6 +14,7 @@ "compressible": "^2.0.18", "debug": "^2.6.9", "negotiator": "^0.6.4", + "on-finished": "^2.4.1", "on-headers": "^1.0.2", "vary": "^1.1.2" }, diff --git a/test/compression.js b/test/compression.js index 449c6198..9ba4838d 100644 --- a/test/compression.js +++ b/test/compression.js @@ -3,6 +3,7 @@ var assert = require('assert') var bytes = require('bytes') var crypto = require('crypto') var http = require('http') +var net = require('net') var request = require('supertest') var zlib = require('zlib') var http2 = require('http2') @@ -953,6 +954,198 @@ describe('compression()', function () { .expect(200, done) }) }) + + describe('when the client closes the connection before consuming the response', function () { + it('should call the original res.end() if connection is cut early on', function (done) { + var server = http.createServer(function (req, res) { + var originalResEnd = res.end + var originalResEndCalledTimes = 0 + res.end = function () { + originalResEndCalledTimes++ + return originalResEnd.apply(this, arguments) + } + + compression({ threshold: 0 })(req, res, function () { + socket.end() + + res.setHeader('Content-Type', 'text/plain') + res.write('hello, ') + setTimeout(function () { + res.end('world!') + + setTimeout(function () { + server.close(function () { + if (originalResEndCalledTimes === 1) { + done() + } else { + done(new Error('The original res.end() was called ' + originalResEndCalledTimes + ' times')) + } + }) + }, 5) + }, 5) + }) + }) + + server.listen() + + var port = server.address().port + var socket = openSocketWithRequest(port) + }) + + it('should call the original res.end() if connection is cut right after setting headers', function (done) { + var server = http.createServer(function (req, res) { + var originalResEnd = res.end + var originalResEndCalledTimes = 0 + res.end = function () { + originalResEndCalledTimes++ + return originalResEnd.apply(this, arguments) + } + + compression({ threshold: 0 })(req, res, function () { + res.setHeader('Content-Type', 'text/plain') + socket.end() + + res.write('hello, ') + setTimeout(function () { + res.end('world!') + + setTimeout(function () { + server.close(function () { + if (originalResEndCalledTimes === 1) { + done() + } else { + done(new Error('The original res.end() was called ' + originalResEndCalledTimes + ' times')) + } + }) + }, 5) + }, 5) + }) + }) + + server.listen() + + var port = server.address().port + var socket = openSocketWithRequest(port) + }) + + it('should call the original res.end() if connection is cut after an initial write', function (done) { + var server = http.createServer(function (req, res) { + var originalResEnd = res.end + var originalResEndCalledTimes = 0 + res.end = function () { + originalResEndCalledTimes++ + return originalResEnd.apply(this, arguments) + } + + compression({ threshold: 0 })(req, res, function () { + res.setHeader('Content-Type', 'text/plain') + res.write('hello, ') + socket.end() + + setTimeout(function () { + res.end('world!') + + setTimeout(function () { + server.close(function () { + if (originalResEndCalledTimes === 1) { + done() + } else { + done(new Error('The original res.end() was called ' + originalResEndCalledTimes + ' times')) + } + }) + }, 5) + }, 5) + }) + }) + + server.listen() + + var port = server.address().port + var socket = openSocketWithRequest(port) + }) + + it('should call the original res.end() if connection is cut just after response body was generated', function (done) { + var server = http.createServer(function (req, res) { + var originalResEnd = res.end + var originalResEndCalledTimes = 0 + res.end = function () { + originalResEndCalledTimes++ + return originalResEnd.apply(this, arguments) + } + + compression({ threshold: 0 })(req, res, function () { + res.setHeader('Content-Type', 'text/plain') + res.write('hello, ') + res.end('world!') + socket.end() + + setTimeout(function () { + server.close(function () { + if (originalResEndCalledTimes === 1) { + done() + } else { + done(new Error('The original res.end() was called ' + originalResEndCalledTimes + ' times')) + } + }) + }, 5) + }) + }) + + server.listen() + + var port = server.address().port + var socket = openSocketWithRequest(port) + }) + + it('should not trigger write errors if connection is cut just after response body was generated', function (done) { + var requestCount = 0 + + var server = http.createServer(function (req, res) { + requestCount += 1 + + var originalWrite = res.write + var writeError = null + res.write = function (chunk, callback) { + return originalWrite.call(this, chunk, function (error) { + if (error) { + writeError = error + } + return callback?.(error) + }) + } + + var originalResEnd = res.end + res.end = function () { + setTimeout(function () { + if (writeError !== null) { + server.close(function () { + done(new Error(`Write error occurred: ${writeError}`)) + }) + } else { + if (requestCount < 50) { + socket = openSocketWithRequest(port) + } else { + server.close(done) + } + } + }, 0) + return originalResEnd.apply(this, arguments) + } + + compression({ threshold: 0 })(req, res, function () { + res.setHeader('Content-Type', 'text/plain') + res.write('hello, ') + res.end('world!') + socket.end() + }) + }) + + server.listen() + + var port = server.address().port + var socket = openSocketWithRequest(port) + }) + }) }) function createServer (opts, fn) { @@ -1056,3 +1249,16 @@ function unchunk (encoding, onchunk, onend) { stream.on('end', onend) } } + +function openSocketWithRequest (port) { + var socket = net.connect(port, function onConnect () { + socket.write('GET / HTTP/1.1\r\n') + socket.write('Accept-Encoding: gzip\r\n') + socket.write('Host: localhost:' + port + '\r\n') + socket.write('Content-Type: text/plain\r\n') + socket.write('Content-Length: 0\r\n') + socket.write('Connection: keep-alive\r\n') + socket.write('\r\n') + }) + return socket +}