Backport fix from v2: Do not swallow calls to res.end()

index.js

@@ -20,6 +20,7 @@ var bytes = require('bytes')
 var compressible = require('compressible')
 var debug = require('debug')('compression')
 var destroy = require('destroy')
+var onFinished = require('on-finished')
 var onHeaders = require('on-headers')
 var vary = require('vary')
 var zlib = require('zlib')
@@ -84,6 +85,7 @@ function compression (options) {
     var listeners = []
     var stream
     var closed = false
+    var originalEnded = false
 
     var _end = res.end
     var _on = res.on
@@ -169,6 +171,13 @@ function compression (options) {
     _on.call(res, 'close', function onResponseClose () {
       closed = true
       destroy(stream)
+
+      // a destroyed stream never emits `end`, so make sure the original
+      // `res.end()` still gets called
+      if (stream && !originalEnded) {
+        originalEnded = true
+        _end.call(res)
+      }
     })
 
     onHeaders(res, function onResponseHeaders () {
@@ -248,18 +257,34 @@ function compression (options) {
 
       // compression
       stream.on('data', function onStreamData (chunk) {
+        if (onFinished.isFinished(res)) {
+          debug('response finished')
+          return
+        }
         if (_write.call(res, chunk) === false) {
+          debug('pausing compression stream')
           stream.pause()
         }
       })
 
       stream.on('end', function onStreamEnd () {
+        if (originalEnded) return
+        originalEnded = true
         _end.call(res)
       })
 
       _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()`.
+      onFinished(res, function onResponseFinished () {
+        stream.resume()
+      })
     })
 
     next()

package.json

@@ -19,6 +19,7 @@
     "debug": "2.6.9",
     "destroy": "1.2.0",
     "negotiator": "~0.6.4",
+    "on-finished": "^2.4.1",
     "on-headers": "~1.1.0",
     "safe-buffer": "5.2.1",
     "vary": "~1.1.2"

test/compression.js

@@ -4,6 +4,7 @@ var Buffer = require('safe-buffer').Buffer
 var bytes = require('bytes')
 var crypto = require('crypto')
 var http = require('http')
+var net = require('net')
 var request = require('supertest')
 var zlib = require('zlib')
 
@@ -1058,6 +1059,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) {
@@ -1241,3 +1434,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
+}