Merge pull request #255 from dead-horse/stream-body

make sure all intermediate stream bodies will be destroy

lib/application.js

@@ -208,12 +208,7 @@ function *respond(next) {
 
   // Stream body
   if ('function' == typeof body.pipe) {
-    if (!~body.listeners('error').indexOf(this.onerror)) body.on('error', this.onerror);
+    if (head) return res.end();
-
-    if (head) {
-      if (body.close) body.close();
-      return res.end();
-    }
 
     return body.pipe(res);
   }

lib/response.js

@@ -9,6 +9,9 @@ var status = require('statuses');
 var http = require('http');
 var path = require('path');
 var mime = require('mime');
+var onFinished = require('finished');
+var destroy = require('dethroy');
+var inject = require('error-inject');
 var basename = path.basename;
 var extname = path.extname;
 
@@ -129,6 +132,12 @@ module.exports = {
 
     // stream
     if ('function' == typeof val.pipe) {
+      // make sure all intermediate streams will destroy on finished
+      onFinished(this, destroy.bind(null, val));
+
+      // make sure all intermediate streams has an listenner on error
+      inject(val, this.ctx.onerror);
+
       if (setType) this.type = 'bin';
       return;
     }

package.json

@@ -33,6 +33,8 @@
     "koa-compose": "~2.2.0",
     "cookies": "~0.4.0",
     "delegates": "0.0.3",
+    "dethroy": "~1.0.0",
+    "error-inject": "~1.0.0",
     "only": "0.0.2"
   },
   "devDependencies": {

test/application.js

@@ -261,7 +261,7 @@ describe('app.respond', function(){
         res.status = 200;
         res.write('Hello');
         setTimeout(function(){
-          res.end("Goodbye")
+          res.end("Goodbye");
         }, 0);
       });
 
@@ -522,6 +522,38 @@ describe('app.respond', function(){
       .expect(404)
       .end(done);
     })
+
+    it('should handle errors when no content status', function(done){
+      var app = koa();
+
+      app.use(function *(){
+        this.status = 204;
+        this.body = fs.createReadStream('does not exist');
+      });
+
+      var server = app.listen();
+
+      request(server)
+      .get('/')
+      .expect(204, done);
+    })
+
+
+    it('should handle all intermediate stream body errors', function(done){
+      var app = koa();
+
+      app.use(function *(){
+        this.body = fs.createReadStream('does not exist');
+        this.body = fs.createReadStream('does not exist');
+        this.body = fs.createReadStream('does not exist');
+      });
+
+      var server = app.listen();
+
+      request(server)
+      .get('/')
+      .expect(404, done);
+    })
   })
 
   describe('when .body is an Object', function(){

test/context.js

@@ -1,11 +1,12 @@
 
+var ReadableStream = require('stream').Readable;
 var context = require('../lib/context');
 var request = require('../lib/request');
 var response = require('../lib/response');
 var koa = require('..');
 
 exports = module.exports = function(req, res){
-  req = req || { headers: {} };
+  req = req || { headers: {}, socket: new ReadableStream() };
   res = res || { _headers: {} };
   res.setHeader = function(k, v){ res._headers[k.toLowerCase()] = v };
   res.removeHeader = function(k, v){ delete res._headers[k.toLowerCase()] };