Update createReadStream() to support streams2 API

Previous support was improperly reverting internal stream to legacy API
causing unnecessary buffering.

Author: lsegal

features/s3/step_definitions/objects.js

@@ -92,7 +92,9 @@ module.exports = function () {
     this.result = '';
     var stream = this.service.getObject(params).createReadStream();
     stream.on('end', function() { callback(); });
-    stream.on('readable', function() { world.result += stream.read(); });
+    stream.on('readable', function() {
+      var v = stream.read(); if (v) world.result += v;
+    });
   });
 
   this.Then(/^the streamed data should contain "([^"]*)"$/, function(data, callback) {

lib/event_listeners.js

@@ -158,16 +158,26 @@ AWS.EventListeners = {
 
     add('SEND', 'send', function SEND(resp) {
       function callback(httpResp) {
+        resp.httpStream = httpResp;
+
         var headers = [httpResp.statusCode, httpResp.headers, resp];
         resp.request.emitEvent('httpHeaders', headers);
 
-        httpResp.on('data', function onData(data) {
-          resp.request.emitEvent('httpData', [data, resp]);
-        });
-
-        httpResp.on('end', function onEnd() {
-          resp.request.emitEvent('httpDone', [resp]);
-        });
+        if (resp.httpStream) {
+          if (AWS.HttpClient.streamsApiVersion === 2) { // streams2 API check
+            httpResp.on('readable', function onReadable() {
+              resp.request.emitEvent('httpData', [httpResp.read(), resp]);
+            });
+          } else { // legacy streams API
+            httpResp.on('data', function onData(data) {
+              resp.request.emitEvent('httpData', [data, resp]);
+            });
+          }
+
+          httpResp.on('end', function onEnd() {
+            resp.request.emitEvent('httpDone', [resp]);
+          });
+        }
       }
 
       function error(err) {
@@ -194,7 +204,7 @@ AWS.EventListeners = {
 
     add('HTTP_DONE', 'httpDone', function HTTP_DONE(resp) {
       // convert buffers array into single buffer
-      if (resp.httpResponse.buffers.length > 0) {
+      if (resp.httpResponse.buffers && resp.httpResponse.buffers.length > 0) {
         var body = AWS.util.buffer.concat(resp.httpResponse.buffers);
         resp.httpResponse.body = body;
       }

lib/http.js

@@ -244,6 +244,11 @@ AWS.NodeHttpClient = inherit({
  */
 AWS.HttpClient = AWS.NodeHttpClient;
 
+/**
+ * @api private
+ */
+AWS.HttpClient.streamsApiVersion = require('stream').Readable ? 2 : 1;
+
 /**
  * @api private
  */

lib/request.js

@@ -270,52 +270,54 @@ AWS.Request = inherit({
    */
   createReadStream: function createReadStream() {
     var req = this;
-    var legacyStreams = !streams.Readable;
-    var stream;
+    var stream = null;
+    var legacyStreams = false;
 
-    if (legacyStreams) {
+    if (AWS.HttpClient.streamsApiVersion === 2) {
+      stream = new streams.Readable();
+      stream._read = function() { stream.push(''); };
+    } else {
       stream = new streams.Stream();
       stream.readable = true;
-    } else {
-      stream = new streams.Readable();
-      stream._read = function() { };
     }
 
+    stream.sent = false;
     stream.on('newListener', function(event) {
-      if ((legacyStreams && event === 'data') ||
-          (!legacyStreams && event == 'readable')) {
+      if (!stream.sent && (event === 'data' || event === 'readable')) {
+        if (event === 'data') legacyStreams = true;
+        stream.sent = true;
         process.nextTick(function() { req.send(); });
       }
     });
 
-    this.on('httpHeaders', function(statusCode) {
+    this.on('httpHeaders', function streamHeaders(statusCode, headers, resp) {
       if (statusCode < 300) {
-        // Remove httpData, httpError listeners, add streaming listeners
         req.removeListener('httpData', AWS.EventListeners.Core.HTTP_DATA);
         req.removeListener('httpError', AWS.EventListeners.Core.HTTP_ERROR);
+        req.on('httpError', function streamHttpError(error, resp) {
+          resp.error = error;
+          resp.error.retryable = false;
+          this.completeRequest(resp);
+        });
 
-        req.on('httpData', function(data) {
-          if (legacyStreams) {
-            stream.emit('data', data);
-          } else {
+        var httpStream = resp.httpStream;
+        stream.response = resp;
+        stream._read = function() {
+          var data;
+          /*jshint boss:true*/
+          while (data = httpStream.read()) {
             stream.push(data);
           }
-        });
+        };
 
-        req.on('httpDone', function() {
-          if (legacyStreams) {
-            stream.emit('end');
-            stream.readable = false;
-          } else {
-            stream.push(null);
-          }
+        var events = ['end', 'error', (legacyStreams ? 'data' : 'readable')];
+        AWS.util.arrayEach(events, function(event) {
+          httpStream.on(event, function(arg) {
+            stream.emit(event, arg);
+          });
         });
 
-        req.on('httpError', function(error, resp) {
-          resp.error = error;
-          resp.error.retryable = false;
-          this.completeRequest(resp);
-        });
+        resp.httpStream = null; // take ownership of the stream object
       }
     });
 

test/helpers.coffee

@@ -59,16 +59,25 @@ MockService = AWS.Service.defineService 'mockService',
     signatureVersion: 'v4'
 
 mockHttpSuccessfulResponse = (status, headers, data, cb) ->
+  if !Array.isArray(data)
+    data = [data]
+
   httpResp = new EventEmitter()
   httpResp.statusCode = status
   httpResp.headers = headers
+  httpResp.read = ->
+    if data.length > 0
+      new Buffer(data.shift())
+    else
+      null
 
   cb(httpResp)
 
-  if !Array.isArray(data)
-    data = [data]
-  AWS.util.arrayEach data, (str) ->
-    httpResp.emit('data', new Buffer(str))
+  AWS.util.arrayEach data.slice(), (str) ->
+    if httpResp._events.readable
+      httpResp.emit('readable')
+    else
+      httpResp.emit('data', new Buffer(str))
 
   httpResp.emit('end')
 

test/request.spec.coffee

@@ -33,6 +33,29 @@ describe 'AWS.Request', ->
       runs ->
         expect(data).toEqual('FOOBARBAZQUX')
 
+    it 'streams2 data (readable event)', ->
+      if AWS.HttpClient.streamsApiVersion < 2
+        return
+
+      data = ''; done = false
+      helpers.mockHttpResponse 200, {}, ['FOO', 'BAR', 'BAZ', 'QUX']
+
+      runs ->
+        request = service.makeRequest('mockMethod')
+        s = request.createReadStream()
+        s.on 'end', -> done = true
+        s.on 'readable', ->
+          try
+            chunk = s.read()
+            if chunk
+              data += chunk
+          catch e
+            console.log(e.stack)
+
+      waitsFor -> done == true
+      runs ->
+        expect(data).toEqual('FOOBARBAZQUX')
+
     it 'does not stream data on failures', ->
       data = ''; error = null; done = false
       helpers.mockHttpResponse 404, {}, ['No such file']