Do not swallow calls to res.end()

index.js

@@ -19,6 +19,7 @@ var Buffer = require('safe-buffer').Buffer
 var bytes = require('bytes')
 var compressible = require('compressible')
 var debug = require('debug')('compression')
+var onFinished = require('on-finished')
 var onHeaders = require('on-headers')
 var vary = require('vary')
 var zlib = require('zlib')
@@ -66,6 +67,14 @@ function compression (options) {
     var _on = res.on
     var _write = res.write
 
+    var endCalled = false
+    function endOnce () {
+      if (!endCalled) {
+        endCalled = true
+        _end.apply(this, arguments)
+      }
+    }
+
     // flush
     res.flush = function flush () {
       if (stream) {
@@ -104,12 +113,16 @@ function compression (options) {
       }
 
       if (!stream) {
-        return _end.call(this, chunk, encoding)
+        return endOnce.call(this, chunk, encoding)
       }
 
       // mark ended
       ended = true
 
+      if (onFinished.isFinished(this)) {
+        return endOnce.call(this)
+      }
+
       // write Buffer for Node.js 0.8
       return chunk
         ? stream.end(toBuffer(chunk, encoding))
@@ -209,12 +222,18 @@ function compression (options) {
       })
 
       stream.on('end', function onStreamEnd () {
-        _end.call(res)
+        endOnce.call(res)
       })
 
       _on.call(res, 'drain', function onResponseDrain () {
         stream.resume()
       })
+
+      onFinished(res, function onFinished () {
+        if (ended) {
+          endOnce.call(res)
+        }
+      })
     })
 
     next()

package.json

@@ -13,6 +13,7 @@
     "bytes": "3.0.0",
     "compressible": "~2.0.18",
     "debug": "2.6.9",
+    "on-finished": "~2.3.0",
     "on-headers": "~1.0.2",
     "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')
 
@@ -657,6 +658,113 @@ describe('compression()', function () {
         .end()
     })
   })
+
+  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 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)
+    })
+  })
 })
 
 function createServer (opts, fn) {
@@ -716,3 +824,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
+}