fix: should not destroy streams

use black-hole-stream to make sure stream's data has been read

Author: dead-horse

lib/response.js

@@ -6,6 +6,7 @@
  */
 
 var contentDisposition = require('content-disposition');
+var BlackHoleStream = require('black-hole-stream');
 var ensureErrorHandler = require('error-inject');
 var getType = require('mime-types').contentType;
 var onFinish = require('on-finished');
@@ -15,6 +16,8 @@ var typeis = require('type-is').is;
 var statuses = require('statuses');
 var destroy = require('destroy');
 var assert = require('assert');
+var Stream = require('stream');
+var http = require('http');
 var path = require('path');
 var vary = require('vary');
 var extname = path.extname;
@@ -161,8 +164,15 @@ module.exports = {
     }
 
     // stream
-    if ('function' == typeof val.pipe) {
-      onFinish(this.res, destroy.bind(null, val));
+    if (val instanceof Stream) {
+      onFinish(this.res, function(){
+        // don't destroy http IncomingMessage, keep `keep-alive` conncetion alive.
+        if (val instanceof http.IncomingMessage) {
+          if (val.readable) val.pipe(new BlackHoleStream());
+        } else {
+          destroy(val);
+        }
+      });
       ensureErrorHandler(val, this.ctx.onerror);
 
       // overwriting

package.json

@@ -19,6 +19,7 @@
   "license": "MIT",
   "dependencies": {
     "accepts": "^1.2.2",
+    "black-hole-stream": "0.0.1",
     "co": "^4.4.0",
     "composition": "^2.1.1",
     "content-disposition": "~0.5.0",
@@ -43,14 +44,18 @@
     "vary": "^1.0.0"
   },
   "devDependencies": {
+    "agentkeepalive": "~2.0.3",
     "babel": "^5.0.0",
+    "freeport": "^1.0.5",
     "istanbul": "^0.4.0",
     "make-lint": "^1.0.1",
     "mocha": "^2.0.1",
+    "pedding": "^1.0.0",
     "should": "^6.0.3",
     "should-http": "0.0.3",
     "supertest": "^1.0.1",
-    "test-console": "^0.7.1"
+    "test-console": "^0.7.1",
+    "urllib": "^2.6.0"
   },
   "engines": {
     "node": ">= 0.12.0",

test/application.js

@@ -2,9 +2,13 @@
 'use strict';
 
 var stderr = require('test-console').stderr;
+var Agent = require('agentkeepalive');
 var request = require('supertest');
 var statuses = require('statuses');
+var freeport = require('freeport');
+var pedding = require('pedding');
 var assert = require('assert');
+var urllib = require('urllib');
 var http = require('http');
 var koa = require('..');
 var fs = require('fs');
@@ -872,6 +876,69 @@ describe('app.respond', function(){
       .get('/')
       .expect(404, done);
     })
+
+    it('should ensure stream do not leak', function(done){
+      done = pedding(3, done);
+      var app = koa();
+      let stream1 = fs.createReadStream(__filename);
+      let stream2 = fs.createReadStream(__filename);
+      stream1.once('close', done);
+      stream2.once('close', done);
+
+      app.use(function *(){
+        this.body = stream1;
+        this.body = stream2;
+      });
+
+      var server = app.listen();
+
+      request(server)
+      .head('/')
+      .expect(200, done);
+    })
+  })
+
+  describe('when .body is a http keepalive IncomingMessage', function(){
+    var target;
+    var port;
+    before(function(done){
+      var app = koa();
+      app.use(function *(){
+        this.body = fs.createReadStream(__filename);
+      });
+
+      freeport(function(err, p){
+        port = p || 12384;
+        target = app.listen(port, done);
+      });
+    })
+
+    after(function(){
+      target.close();
+    })
+
+    it('should not destroy keepalive connection', function(done){
+      done = pedding(2, done);
+      var app = koa();
+      app.use(function *(){
+        var remote = yield urllib.request('http://127.0.0.1:' + port, {
+          streaming: true,
+          agent: new Agent()
+        });
+        var res = remote.res;
+        this.body = res;
+        res.once('end', function(){
+          assert.equal(res.readable, false);
+          assert.equal(res.socket.destroyed, false);
+          done();
+        });
+      });
+
+      var server = app.listen();
+      request(server)
+      .head('/')
+      .expect(200, done);
+    })
   })
 
   describe('when .body is an Object', function(){