request: add support for streaming bodies

Author: aredridel

lib/request.js

@@ -146,13 +146,39 @@ function makeRequest (uri, params, cb_) {
   this.log.http('request', params.method, parsed.href || '/')
 
   var done = requestDone.call(this, params.method, uri, cb)
-  var req = request(opts, decodeResponseBody(done))
+  var req = request(opts, params.streaming ? undefined : decodeResponseBody(done))
 
   req.on('error', cb)
   req.on('socket', function (s) {
     s.on('error', cb)
   })
 
+  if (params.streaming) {
+    req.on('response', function (response) {
+
+      if (response.statusCode >= 400) {
+        var parts = []
+        response.on('data', function (data) {
+          parts.push(data)
+        })
+        response.on('end', function () {
+          decodeResponseBody(done)(null, response, Buffer.concat(parts))
+        })
+      } else {
+
+        response.on('end', function () {
+          // don't ever re-use connections that had server errors.
+          // those sockets connect to the Bad Place!
+          if (response.socket && response.statusCode > 500) {
+            response.socket.destroy()
+          }
+        })
+
+        return cb(null, response)
+      }
+    })
+  }
+
   if (params.body && (params.body instanceof Stream)) {
     params.body.pipe(req)
   }

test/fetch-streaming.js

@@ -0,0 +1,59 @@
+var test = require('tap').test
+var concat = require('concat-stream')
+
+var server = require('./lib/server.js')
+var common = require('./lib/common.js')
+var client = common.freshClient()
+
+var testData = JSON.stringify({test: true})
+var errorData = JSON.stringify({error: 'it went bad'})
+
+test('streaming fetch', function (t) {
+  server.expect('/test', function (req, res) {
+    t.equal(req.method, 'GET', 'got expected method')
+
+    res.writeHead(200, {
+      'content-type': 'application/json'
+    })
+
+    res.end(testData)
+  })
+
+  server.expect('/error', function (req, res) {
+    t.equal(req.method, 'GET', 'got expected method')
+
+    res.writeHead(401, {
+      'content-type': 'application/json'
+    })
+
+    res.end(errorData)
+  })
+
+  client.fetch(
+    'http://localhost:1337/test',
+    { streaming: true },
+    function (er, res) {
+      t.ifError(er, 'loaded successfully')
+
+      var sink = concat(function (data) {
+        t.deepEqual(data.toString(), testData)
+        client.fetch(
+          'http://localhost:1337/error',
+          { streaming: true },
+          function (er, res) {
+            t.ok(er, 'got an error')
+            server.close()
+            t.end()
+          }
+        )
+
+      })
+
+      res.on('error', function (error) {
+        t.ifError(error, 'no errors on stream')
+      })
+
+      res.pipe(sink)
+    }
+  )
+})