增加对 stream 的支持

README.md

@@ -61,7 +61,7 @@ These are the available config options for making requests. Only the url is requ
   timeout: 1000, // default is `15000` milliseconds
 
   // `responseType` indicates the type of data that the server will respond with
-  // options are: 'json', 'text', 'buffer'
+  // options are: 'json', 'text', 'buffer'，'stream'
   responseType: 'json', // default
 
   // `agent` define a custom agent to be used when performing http or https requests,

lib/request.js

@@ -2,6 +2,7 @@ const http = require('http');
 const https = require('https');
 const { URL } = require('url');
 const qs = require('querystring');
+const pump = require('pump');
 const Response = require('./response');
 const createError = require('./createError');
 const enhanceError = require('./enhanceError');
@@ -41,12 +42,20 @@ async function request(options = {}) {
     options.headers || {}
   );
   const contentType = headers['Content-Type'];
+  const isStream =
+    data &&
+    data.readable !== false &&
+    typeof data === 'object' &&
+    typeof data.pipe === 'function' &&
+    typeof data._read === 'function' &&
+	  typeof data._readableState === 'object'
+
   if (
     Object.prototype.toString.call(data) === '[object Object]' &&
     contentType.startsWith('application/x-www-form-urlencoded')
   ) {
     data = qs.stringify(data);
-  } else if (Object.prototype.toString.call(data) !== '[object String]') {
+  } else if (Object.prototype.toString.call(data) !== '[object String]' && !isStream) {
     data = JSON.stringify(data);
   }
 
@@ -67,12 +76,20 @@ async function request(options = {}) {
 
     const resHandler = (res) => {
       const statusCode = res.statusCode;
+
       let response = new Response({
         headers: res.headers,
         statusCode,
         statusMessage: res.statusMessage,
       });
 
+      // 如果 responseType 为 stream 直接返回
+      if (options.responseType === 'stream') {
+        response.data = res;
+        resolve(response);
+        return;
+      }
+
       res.on('data', (chunk) => {
         response.addChunk(chunk);
       });
@@ -91,6 +108,13 @@ async function request(options = {}) {
       });
     };
 
+    const errHandler = (err) => {
+      if (req.aborted) {
+        return;
+      }
+      reject(enhanceError(err, options, null, req));
+    }
+
     let req;
     if (protocol === 'http:') {
       req = http.request(requestOptions, resHandler);
@@ -101,24 +125,27 @@ async function request(options = {}) {
     }
 
     // Handle errors
-    req.on('error', (err) => {
-      if (req.aborted) {
-        return;
-      }
-      reject(enhanceError(err, options, null, req));
-    });
+    req.on('error', errHandler);
 
     // Handle request timeout
     if (timeout) {
       req.setTimeout(timeout, () => {
         req.abort();
+
+        if (isStream) {
+          data.destroy();
+        }
+
         reject(createError('timeout of ' + timeout + 'ms exceeded', options, 'ECONNABORTED', req));
       });
     }
 
-    if (data && ['GET', 'HEAD'].indexOf(method) === -1) {
+    if (isStream) {
+      return pump(data, req, (err) => err && errHandler(err));
+    } else if (data && ['GET', 'HEAD'].indexOf(method) === -1) {
       req.write(data);
     }
+
     req.end();
   });
 }

package.json

@@ -11,7 +11,7 @@
   },
   "ava": {
     "files": [
-      "test/**/*.js"
+      "test/**/*.test.js"
     ]
   },
   "repository": {
@@ -30,6 +30,7 @@
   },
   "homepage": "https://github.com/node-labx/lightning-request",
   "dependencies": {
+    "pump": "^3.0.0",
     "sync-rpc": "^1.3.6"
   }
 }

test/file/tmp.txt

@@ -0,0 +1 @@
+this is a text file for stream test

test/request-with-stream.test.js

@@ -0,0 +1,64 @@
+const test = require('ava');
+const fs = require('fs');
+const path = require('path');
+const request = require('../index');
+const httpServer = require('./server/index');
+
+let port;
+let server;
+
+test.before('run server', async () => {
+  try {
+    const result = await httpServer();
+    port = result.port;
+    server = result.server;
+  } catch (err) {
+    console.log('test before hook error: ', err);
+  }
+});
+
+test.after('close server', function() {
+  server && server.close();
+});
+
+test('put request with stream', async (t) => {
+  const tmp = path.join(process.cwd(), 'test/file/tmp.txt');
+  const stream = fs.createReadStream(tmp);
+
+  try {
+    const result = await request({
+      url: 'http://localhost:' + port + '/stream',
+      data: stream,
+      method: 'PUT',
+    });
+
+    t.is(result.data, fs.readFileSync(tmp, { encoding: 'utf-8'}));
+  } catch (err) {
+    console.log(err);
+    t.fail(err);
+  }
+});
+
+test.cb('get request with stream responseType', (t) => {
+  t.plan(1);
+
+  request({
+    url: 'http://localhost:' + port + '/stream',
+    responseType: 'stream',
+  }).then((result) => {
+    const chunks  = [];
+
+    result.data.on('data', (buf) => chunks.push(buf));
+    result.data.on('end', () => {
+      const datas = Buffer.concat(chunks).toString();
+      const tmp = path.join(process.cwd(), 'test/file/tmp.txt');
+
+      t.is(datas, fs.readFileSync(tmp, { encoding: 'utf-8'}));
+      t.end();
+    })
+  }).catch((err) => {
+    console.log(err);
+    t.fail(err);
+    t.end();
+  });
+})

test/server/index.js

@@ -0,0 +1,47 @@
+const http = require('http');
+const fs = require('fs');
+const path = require('path');
+
+module.exports = () => {
+  return new Promise((resolve, reject) => {
+    const server = http.createServer(function(req, res) {
+      const chunks  = [];
+
+      req.on('data', (buf) => {
+        chunks.push(buf);
+      });
+
+      req.on('end', () => {
+        const content = Buffer.concat(chunks).toString();
+
+        if (req.url === '/stream') {
+          if (req.method === 'PUT') {
+            res.end(content);
+          } else if (req.method === 'GET') {
+            const tmp = path.join(process.cwd(), 'test/file/tmp.txt');
+            res.end(fs.readFileSync(tmp, { encoding: 'utf-8' }))
+          }
+
+          return;
+        }
+
+        res.end('test server')
+      });
+    });
+
+    server.on('error', reject);
+    server.on('close', () => {
+      // console.log('test server closeed');
+    })
+
+    server.listen(function() {
+      const address = server.address();
+
+      // console.log('test server listen at ', address);
+      resolve({
+        port: address.port,
+        server,
+      });
+    });
+  })
+};