fix: should consume the response stream on error (ali-sdk#503)

.travis.yml

@@ -1,7 +1,7 @@
 sudo: false
 language: node_js
 node_js:
-  - '9'
+  - '10'
   - '8'
 env:
   global:

lib/client.js

@@ -1,12 +1,14 @@
 
 const debug = require('debug')('ali-oss');
+const sendToWormhole = require('stream-wormhole');
 const crypto = require('crypto');
 const path = require('path');
 const copy = require('copy-to');
 const mime = require('mime');
 const xml = require('xml2js');
 const ms = require('humanize-ms');
 const AgentKeepalive = require('agentkeepalive');
+const HttpsAgentKeepalive = require('agentkeepalive').HttpsAgent;
 const merge = require('merge-descriptors');
 const urlutil = require('url');
 const is = require('is-type-of');
@@ -20,6 +22,7 @@ const signUtils = require('./common/signUtils');
 const utils = require('./common/utils');
 
 const globalHttpAgent = new AgentKeepalive();
+const globalHttpsAgent = new HttpsAgentKeepalive();
 
 function getHeader(headers, name) {
   return headers[name] || headers[name.toLowerCase()];
@@ -68,6 +71,7 @@ function Client(options, ctx) {
   } else {
     this.urllib = urllib;
     this.agent = this.options.agent || globalHttpAgent;
+    this.httpsAgent = this.options.httpsAgent || globalHttpsAgent;
   }
   this.ctx = ctx;
   this.userAgent = this._getUserAgent();
@@ -254,7 +258,6 @@ proto.createRequest = function createRequest(params) {
   debug('request %s %s, with headers %j, !!stream: %s', params.method, url, headers, !!params.stream);
   const timeout = params.timeout || this.options.timeout;
   const reqParams = {
-    agent: this.agent,
     method: params.method,
     content: params.content,
     stream: params.stream,
@@ -264,6 +267,12 @@ proto.createRequest = function createRequest(params) {
     customResponse: params.customResponse,
     ctx: params.ctx || this.ctx,
   };
+  if (this.agent) {
+    reqParams.agent = this.agent;
+  }
+  if (this.httpsAgent) {
+    reqParams.httpsAgent = this.httpsAgent;
+  }
 
   return {
     url,
@@ -309,6 +318,10 @@ proto.request = async function request(params) {
   }
 
   if (err) {
+    if (params.customResponse && result && result.res) {
+      // consume the response stream
+      await sendToWormhole(result.res);
+    }
     throw err;
   }
 

package.json

@@ -100,7 +100,8 @@
   },
   "dependencies": {
     "address": "^1.0.0",
-    "agentkeepalive": "^2.1.1",
+    "agentkeepalive": "^3.4.1",
+    "any-promise": "^1.3.0",
     "bowser": "^1.6.0",
     "co-defer": "^1.0.0",
     "copy-to": "^2.0.1",
@@ -114,12 +115,13 @@
     "jstoxml": "^0.2.3",
     "merge-descriptors": "^1.0.1",
     "mime": "^1.3.4",
+    "mz-modules": "^2.1.0",
     "platform": "^1.3.1",
     "sdk-base": "^2.0.1",
+    "stream-http": "2.8.2",
+    "stream-wormhole": "^1.0.4",
     "urllib": "^2.17.1",
     "utility": "^1.8.0",
-    "xml2js": "^0.4.16",
-    "stream-http": "2.8.2",
-    "any-promise": "^1.3.0"
+    "xml2js": "^0.4.16"
   }
 }

test/node/object.test.js

@@ -3,6 +3,9 @@ const fs = require('fs');
 const path = require('path');
 const assert = require('assert');
 const { Readable } = require('stream');
+const AgentKeepalive = require('agentkeepalive');
+const HttpsAgentKeepalive = require('agentkeepalive').HttpsAgent;
+const sleep = require('mz-modules/sleep');
 const utils = require('./utils');
 const oss = require('../..');
 const config = require('../config').oss;
@@ -291,7 +294,7 @@ describe('test/object.test.js', () => {
       assert.equal(info.res.headers['content-type'], 'application/javascript; charset=utf8');
     });
 
-    it('should return correct encode when name include + and space', async  () => {
+    it('should return correct encode when name include + and space', async () => {
       const name = 'ali-sdkhahhhh+oss+mm xxx.js';
       const object = await store.put(name, __filename, {
         headers: {
@@ -971,6 +974,22 @@ describe('test/object.test.js', () => {
         assert.equal(err.name, 'NoSuchKeyError');
       }
     });
+
+    it('should throw error and consume the response stream', async () => {
+      store.agent = new AgentKeepalive({
+        keepAlive: true,
+      });
+      store.httpsAgent = new HttpsAgentKeepalive();
+      try {
+        await store.getStream(`${name}not-exists`);
+        throw new Error('should not run this');
+      } catch (err) {
+        assert.equal(err.name, 'NoSuchKeyError');
+        assert(Object.keys(store.agent.freeSockets).length === 0);
+        await sleep(1);
+        assert(Object.keys(store.agent.freeSockets).length === 1);
+      }
+    });
   });
 
   describe('delete()', () => {