converted decompression to use streams rather than callbacks

src/lib/connectors/http.js

@@ -122,11 +122,10 @@ HttpConnector.prototype.request = function (params, cb) {
   var incoming;
   var timeoutId;
   var request;
-  var response;
   var status = 0;
   var headers = {};
   var log = this.log;
-  var buffers = [];
+  var response;
 
   var reqParams = this.makeReqParams(params);
 
@@ -146,21 +145,7 @@ HttpConnector.prototype.request = function (params, cb) {
     if (err) {
       cb(err);
     } else {
-      response = Buffer.concat(buffers);
-      var zipHdr = headers['content-encoding'];
-      if (zipHdr && (zipHdr.match(/gzip/i) || zipHdr.match(/deflate/i))) {
-        zlib.unzip(response, function(gzErr, uncompressedResponse) {
-          if(gzErr) {
-            err = gzErr;
-            response = response.toString('binary');
-          } else {
-            response = uncompressedResponse.toString('utf8');
-          }
       cb(err, response, status, headers);
-        });
-      } else {
-        cb(err, response.toString('utf8'), status, headers);
-      }
     }
   }, this);
 
@@ -169,9 +154,15 @@ HttpConnector.prototype.request = function (params, cb) {
     status = incoming.statusCode;
     headers = incoming.headers;
     response = '';
-    buffers = [];
+
+    var encoding = (headers['content-encoding'] || '').toLowerCase();
+    if (encoding === 'gzip' || encoding === 'deflate') {
+      incoming = incoming.pipe(zlib.createUnzip());
+    }
+
+    incoming.setEncoding('utf8');
     incoming.on('data', function (d) {
-      buffers.push(new Buffer(d));
+      response += d;
     });
 
     incoming.on('error', cleanUp);

test/mocks/incomming_message.js

@@ -15,12 +15,11 @@ if (!Readable) {
 }
 
 function MockIncommingMessage() {
-  var self = this;
+  Readable.call(this);
 
-  Readable.call(self);
-
-  self.setEncoding = sinon.stub();
-  self._read = function () {};
+  this.setEncoding = sinon.stub();
+  this.headers = {};
+  this._read = function () {};
 }
 util.inherits(MockIncommingMessage, Readable);
 

test/unit/specs/http_connector.js

@@ -308,11 +308,11 @@ describe('Http Connector', function () {
       var server = nock('http://esjs.com:9200');
       var con = new HttpConnection(new Host('http://esjs.com:9200'));
       var elements = [];
-      for(var i = 0; i < 500; i++) {
-        elements.push({ "USER": "doc" });
+      for (var i = 0; i < 500; i++) {
+        elements.push({ USER: 'doc' });
       }
       var body = JSON.stringify(elements);
-      zlib.gzip(body, function(err, compressedBody) {
+      zlib.gzip(body, function (err, compressedBody) {
         server
           .get('/users/1')
           .reply(200, compressedBody, {'Content-Encoding': 'gzip'});
@@ -334,11 +334,11 @@ describe('Http Connector', function () {
       var server = nock('http://esjs.com:9200');
       var con = new HttpConnection(new Host('http://esjs.com:9200'));
       var elements = [];
-      for(var i = 0; i < 500; i++) {
-        elements.push({ "USER": "doc" });
+      for (var i = 0; i < 500; i++) {
+        elements.push({ USER: 'doc' });
       }
       var body = JSON.stringify(elements);
-      zlib.deflate(body, function(err, compressedBody) {
+      zlib.deflate(body, function (err, compressedBody) {
         server
           .get('/users/1')
           .reply(200, compressedBody, {'Content-Encoding': 'deflate'});
@@ -368,9 +368,9 @@ describe('Http Connector', function () {
         method: 'GET',
         path: '/users/1'
       }, function (err, resp, status) {
-        expect(err.errno).to.be(-3);
-        expect(resp).to.eql(body);
-        expect(status).to.eql(200);
+        expect(err).to.be.an(Error);
+        expect(resp).to.eql(undefined);
+        expect(status).to.eql(undefined);
         server.done();
         done();
       });