Support Buffer & Stream payloads in REST services

The "Body" payload property of REST-based services (like Amazon S3)
will now return Buffer objects instead of Strings, which allows for
the downloading of binary data:

```js
s3.client.getObject({Bucket:'foo',Key:'bar'}, function(err, data) {
  console.log(data.Body); // data.Body is now a Buffer object
});
```

In addition, you can now assign a readable Stream object to the
Body payload:

```js
// Read the body from a stream
var writeParams = {Bucket:'bucket',Key:'image.jpg',Body:stream};
var stream = fs.createReadStream('/path/to/image.jpg');
s3.client.putObject(writeParams, function() {
  // Write the body Buffer object to a file
  var readParams = {Bucket:'bucket',Key:'image.jpg'};
  s3.client.getObject(readParams, function(err, data) {
    var out = fs.createWriteStream('/path/to/image2.jpg');
    out.write(data.Body);
    out.end();
  })
});
```

Conflicts:
  lib/service_interface/query.js

Fixes #3

Author: lsegal

features/s3/objects.feature

@@ -33,5 +33,17 @@ Feature: Working with Objects in S3
     When I delete the object with the key "hello"
     Then the object with the key "hello" should not exist
 
+  @buffer
+  Scenario: Buffers and streams
+    When I write buffer "world" to the key "hello"
+    Then the object with the key "hello" should exist
+    And the object with the key "hello" should contain "world"
+    And I delete the object with the key "hello"
+
+    When I write file "testfile.txt" to the key "hello"
+    Then the object with the key "hello" should exist
+    And the object with the key "hello" should contain "CONTENTS OF FILE"
+    And I delete the object with the key "hello"
+
     # final step here needs to happen to cleanup the shared bucket
     And I delete the shared bucket

features/s3/step_definitions/objects.js

@@ -31,18 +31,17 @@ module.exports = function () {
     this.s3.createBucket({Bucket:this.sharedBucket}, function(err, data) {
       callback();
     });
-
   });
 
-  this.When(/^I write "([^"]*)" to the key "([^"]*)"$/, function(contents, key, next) {
-    var params = {Bucket:this.sharedBucket,Key:key,Body:contents};
+  this.When(/^I write (buffer )?"([^"]*)" to the key "([^"]*)"$/, function(buffer, contents, key, next) {
+    var params = {Bucket: this.sharedBucket, Key: key, Body: buffer ? new Buffer(contents) : contents};
     this.request('s3', 'putObject', params, next);
   });
 
   this.Then(/^the object with the key "([^"]*)" should contain "([^"]*)"$/, function(key, contents, next) {
     this.eventually(next, function (retry) {
       this.s3.getObject({Bucket:this.sharedBucket,Key:key}, function(err, data) {
-        if (data && data.Body == contents)
+        if (data && data.Body.toString().replace("\n", "") == contents)
           next();
         else
           retry();
@@ -68,9 +67,15 @@ module.exports = function () {
     });
   });
 
+  this.When(/^I write file "([^"]*)" to the key "([^"]*)"$/, function(filename, key, next) {
+    var fs = require('fs');
+    var params = {Bucket: this.sharedBucket, Key: key, Body:
+      fs.createReadStream(__dirname + '/../../support/fixtures/' + filename)};
+    this.request('s3', 'putObject', params, next);
+  });
+
   // this scenario is a work around for not having an after all hook
   this.Then(/^I delete the shared bucket$/, function(next) {
     this.request('s3', 'deleteBucket', {Bucket:this.sharedBucket}, next);
   });
-
 };

features/support/fixtures/testfile.txt

@@ -0,0 +1 @@
+CONTENTS OF FILE

features/support/helpers.js

@@ -71,14 +71,15 @@ module.exports = {
    * finish execution before moving onto the next step in the scenario.
    */
   request: function request(svc, operation, params, next) {
-    this[svc][operation](params).on('complete', function (resp) {
-      if (resp.error) {
-        this.unexpectedError(resp, next);
+    var world = this;
+    this[svc][operation](params, function(err, data) {
+      if (err) {
+        world.unexpectedError(this, next);
       } else {
-        this.resp = resp;
+        world.resp = this;
         next();
       }
-    }, { bind: this }).send();
+    });
   },
 
   /**
@@ -87,8 +88,8 @@ module.exports = {
    * operation failed.
    */
   unexpectedError: function unexpectedError(resp, next) {
-    var svc = resp.service.serviceName;
-    var op = resp.method;
+    var svc = resp.request.client.serviceName;
+    var op = resp.request.operation;
     var code = resp.error.code;
     var msg = resp.error.message;
     var err = 'Received unexpected error from ' + svc + '.' + op + ', ' + code + ': ' + msg;

lib/event_listeners.js

@@ -82,14 +82,19 @@ AWS.EventListeners = {
         function HTTP_HEADERS(statusCode, headers, resp) {
       resp.httpResponse.statusCode = statusCode;
       resp.httpResponse.headers = headers;
-      resp.httpResponse.body = '';
+      resp.httpResponse.body = new Buffer('');
+      resp.httpResponse.buffers = [];
     });
 
     add('HTTP_DATA', 'httpData', function HTTP_DONE(chunk, resp) {
-      resp.httpResponse.body += chunk;
+      resp.httpResponse.buffers.push(chunk);
     });
 
     add('HTTP_DONE', 'httpDone', function HTTP_DONE(resp) {
+      // convert buffers array into single buffer
+      resp.httpResponse.body = Buffer.concat(resp.httpResponse.buffers);
+      delete resp.httpResponse.buffers;
+
       this.completeRequest(resp);
     });
 

lib/http.js

@@ -14,6 +14,7 @@
  */
 
 var AWS = require('./core');
+var Stream = require('stream').Stream;
 var inherit = AWS.util.inherit;
 
 /**
@@ -172,15 +173,17 @@ AWS.NodeHttpClient = inherit({
       options.agent = new client.Agent(options);
     }
 
-    var req = this.setupEvents(client, options, request, response);
-    if (request.httpRequest.body) {
-      req.write(request.httpRequest.body);
+    var stream = this.setupEvents(client, options, request, response);
+    if (request.httpRequest.body instanceof Stream) {
+      request.httpRequest.body.pipe(stream, {end: false});
+    } else if (request.httpRequest.body) {
+      stream.write(request.httpRequest.body);
     }
-    req.end();
+    stream.end();
   },
 
   setupEvents: function setupEvents(client, options, request, response) {
-    var req = client.request(options, function onResponse(httpResponse) {
+    var stream = client.request(options, function onResponse(httpResponse) {
       request.emit('httpHeaders', httpResponse.statusCode,
                                   httpResponse.headers, response, request);
 
@@ -193,12 +196,12 @@ AWS.NodeHttpClient = inherit({
       });
     });
 
-    req.on('error', function (err) {
+    stream.on('error', function (err) {
       request.emit('httpError', AWS.util.error(err,
         {code: 'NetworkingError', retryable: true}));
     });
 
-    return req;
+    return stream;
   }
 });
 

lib/service_interface/json.js

@@ -31,8 +31,8 @@ AWS.ServiceInterface.Json = {
     var error = {};
     var httpResponse = resp.httpResponse;
 
-    if (httpResponse.body) {
-      var e = JSON.parse(httpResponse.body);
+    if (httpResponse.body.length > 0) {
+      var e = JSON.parse(httpResponse.body.toString());
       error.code = e.__type.split('#').pop();
       if (error.code === 'RequestEntityTooLarge') {
         error.message = 'Request body must be less than 1 MB';
@@ -48,6 +48,6 @@ AWS.ServiceInterface.Json = {
   },
 
   extractData: function extractData(resp) {
-    resp.data = JSON.parse(resp.httpResponse.body || '{}');
+    resp.data = JSON.parse(resp.httpResponse.body.toString() || '{}');
   }
 };

lib/service_interface/query.js

@@ -39,7 +39,7 @@ AWS.ServiceInterface.Query = {
   },
 
   extractError: function extractError(resp) {
-    var data = new AWS.XML.Parser({}).parse(resp.httpResponse.body);
+    var data = new AWS.XML.Parser({}).parse(resp.httpResponse.body.toString());
     if (data.Code) {
       resp.error = AWS.util.error(new Error(), {
         code: data.Code,
@@ -57,7 +57,7 @@ AWS.ServiceInterface.Query = {
     var req = resp.request;
     var operation = req.client.api.operations[req.operation];
     var parser = new AWS.XML.Parser(operation.o || {});
-    resp.data = parser.parse(resp.httpResponse.body);
+    resp.data = parser.parse(resp.httpResponse.body.toString());
   }
 };
 

lib/service_interface/rest_xml.js

@@ -26,7 +26,7 @@ AWS.ServiceInterface.RestXml = {
   extractError: function extractError(resp) {
     AWS.ServiceInterface.Rest.extractError(resp);
 
-    var data = new AWS.XML.Parser({}).parse(resp.httpResponse.body);
+    var data = new AWS.XML.Parser({}).parse(resp.httpResponse.body.toString());
     if (data.Code) {
       resp.error = AWS.util.error(new Error(), {
         code: data.Code,
@@ -48,15 +48,15 @@ AWS.ServiceInterface.RestXml = {
     var operation = req.client.api.operations[req.operation];
     var rules = operation.o || {};
 
-    if (rules['Body'] && rules['Body']['t'] === 'bl') {
-      resp.data['Body'] = httpResponse.body;
-    } else if (httpResponse.body !== '') {
+    if (rules.Body && rules.Body.t === 'bl') {
+      resp.data.Body = httpResponse.body;
+    } else if (httpResponse.body.length > 0) {
       var parser = new AWS.XML.Parser(rules);
-      AWS.util.update(resp.data, parser.parse(httpResponse.body));
+      AWS.util.update(resp.data, parser.parse(httpResponse.body.toString()));
     }
 
     // extract request id
-    resp.data['RequestId'] = httpResponse.headers['x-amz-request-id'];
+    resp.data.RequestId = httpResponse.headers['x-amz-request-id'];
   },
 
   populateBody: function populateBody(req) {
@@ -88,6 +88,16 @@ AWS.ServiceInterface.RestXml = {
     }
 
     req.httpRequest.body = body;
-    req.httpRequest.headers['Content-Length'] = body ? body.length : 0;
+    req.httpRequest.headers['Content-Length'] =
+      AWS.ServiceInterface.RestXml.getBodyLength(body);
+  },
+
+  getBodyLength: function getBodyLength(body) {
+    if (!body) return 0;
+    if (body.length) {
+      return body.length;
+    } else if (body.path) { // assume file system stream
+      return require('fs').lstatSync(body.path).size;
+    }
   }
 };

lib/services/ec2.js

@@ -50,7 +50,7 @@ AWS.EC2.Client = inherit(AWS.Client, {
   extractError: function extractError(resp) {
     // EC2 nests the error code and message deeper than other AWS Query services.
     var httpResponse = resp.httpResponse;
-    var data = new AWS.XML.Parser({}).parse(httpResponse.body || '');
+    var data = new AWS.XML.Parser({}).parse(httpResponse.body.toString() || '');
     if (data.Errors)
       resp.error = AWS.util.error(new Error(), {
         code: data.Errors.Error.Code,

lib/services/s3.js

@@ -126,7 +126,7 @@ AWS.S3.Client = inherit(AWS.Client, {
     var req = resp.request;
     var httpResponse = resp.httpResponse;
     if (req.operation === 'completeMultipartUpload' &&
-        httpResponse.body.match('<Error>'))
+        httpResponse.body.toString().match('<Error>'))
       return false;
     else
       return httpResponse.statusCode < 300;
@@ -156,7 +156,7 @@ AWS.S3.Client = inherit(AWS.Client, {
     var req = resp.request;
     if (req.operation === 'getBucketLocation') {
       /*jshint regexp:false*/
-      var match = resp.httpResponse.body.match(/>(.+)<\/Location/);
+      var match = resp.httpResponse.body.toString().match(/>(.+)<\/Location/);
       if (match) {
         delete resp.data['_'];
         resp.data.LocationConstraint = match[1];
@@ -182,7 +182,7 @@ AWS.S3.Client = inherit(AWS.Client, {
         message: null
       });
     } else {
-      var data = new AWS.XML.Parser({}).parse(resp.httpResponse.body);
+      var data = new AWS.XML.Parser({}).parse(resp.httpResponse.body.toString());
 
       resp.error = AWS.util.error(new Error(), {
         code: data.Code || resp.httpResponse.statusCode,

test/helpers.coffee

@@ -48,7 +48,7 @@ MockClient = AWS.util.inherit AWS.Client,
     @config.region = 'mock-region'
   setupRequestListeners: (request) ->
     request.on 'extractData', (resp) ->
-      resp.data = resp.httpResponse.body
+      resp.data = resp.httpResponse.body.toString()
     request.on 'extractError', (resp) ->
       resp.error =
         code: resp.httpResponse.statusCode
@@ -68,7 +68,7 @@ mockHttpResponse = (status, headers, data) ->
       req.emit('httpHeaders', status, headers, resp)
       str = str instanceof Array ? str : [str]
       AWS.util.arrayEach data, (str) ->
-        req.emit('httpData', str, resp)
+        req.emit('httpData', new Buffer(str), resp)
       req.emit('httpDone', resp)
     else
       req.emit('httpError', status, resp)

test/unit/event_listeners.spec.coffee

@@ -95,7 +95,7 @@ describe 'AWS.EventListeners', ->
 
       # register httpData event
       request = makeRequest()
-      request.on('httpData', (chunk) -> calls.push(chunk))
+      request.on('httpData', (chunk) -> calls.push(chunk.toString()))
       request.send()
 
       expect(calls).toEqual(['FOO', 'BAR', 'BAZ', 'QUX'])
@@ -105,7 +105,7 @@ describe 'AWS.EventListeners', ->
       request.on('httpData', ->)
       response = request.send()
 
-      expect(response.httpResponse.body).toEqual('')
+      expect(response.httpResponse.body.toString()).toEqual('')
 
   describe 'retry', ->
     it 'retries a request with a set maximum retries', ->

test/unit/service_interface/json.spec.coffee

@@ -75,7 +75,7 @@ describe 'AWS.ServiceInterface.Json', ->
   describe 'extractError', ->
     extractError = (body) ->
       response.httpResponse.statusCode = 500
-      response.httpResponse.body = body
+      response.httpResponse.body = new Buffer(body)
       svc.extractError(response)
 
     it 'removes prefixes from the error code', ->
@@ -91,7 +91,7 @@ describe 'AWS.ServiceInterface.Json', ->
       expect(response.data).toEqual(null)
 
     it 'returns the status code when the body is blank', ->
-      extractError null
+      extractError ''
       expect(response.error instanceof Error).toBeTruthy()
       expect(response.error.code).toEqual(500)
       expect(response.error.message).toEqual(null)
@@ -125,7 +125,7 @@ describe 'AWS.ServiceInterface.Json', ->
   describe 'extractData', ->
     extractData = (body) ->
       response.httpResponse.statusCode = 200
-      response.httpResponse.body = body
+      response.httpResponse.body = new Buffer(body)
       svc.extractData(response)
 
     it 'JSON parses http response bodies', ->
@@ -139,6 +139,6 @@ describe 'AWS.ServiceInterface.Json', ->
       expect(response.data).toEqual({})
 
     it 'returns an empty object when the body is null', ->
-      extractData null
+      extractData ''
       expect(response.error).toEqual(null)
       expect(response.data).toEqual({})

test/unit/service_interface/query.spec.coffee

@@ -81,7 +81,7 @@ describe 'AWS.ServiceInterface.Query', ->
         </Error>
         """
       response.httpResponse.statusCode = 400
-      response.httpResponse.body = body
+      response.httpResponse.body = new Buffer(body)
       svc.extractError(response)
 
     it 'extracts the error code and message', ->
@@ -100,7 +100,7 @@ describe 'AWS.ServiceInterface.Query', ->
   describe 'extractData', ->
     extractData = (body) ->
       response.httpResponse.statusCode = 200
-      response.httpResponse.body = body
+      response.httpResponse.body = new Buffer(body)
       svc.extractData(response)
 
     it 'parses the response using the operation output rules', ->

test/unit/service_interface/rest_xml.spec.coffee

@@ -188,7 +188,7 @@ describe 'AWS.ServiceInterface.RestXml', ->
         </Error>
         """
       response.httpResponse.statusCode = 400
-      response.httpResponse.body = body
+      response.httpResponse.body = new Buffer(body)
       svc.extractError(response)
 
     it 'extracts the error code and message', ->
@@ -208,7 +208,7 @@ describe 'AWS.ServiceInterface.RestXml', ->
   describe 'extractData', ->
     extractData = (body) ->
       response.httpResponse.statusCode = 200
-      response.httpResponse.body = body
+      response.httpResponse.body = new Buffer(body)
       svc.extractData(response)
 
     it 'parses the xml body', ->
@@ -224,3 +224,9 @@ describe 'AWS.ServiceInterface.RestXml', ->
       </xml>
       """
       expect(response.data).toEqual({Foo:'foo', Bar:['a', 'b', 'c']})
+
+    it 'sets Body to a Buffer object', ->
+      operation.o = Body: t: 'bl'
+      extractData 'Buffer data'
+      expect(response.data.Body instanceof Buffer).toBeTruthy()
+      expect(response.data.Body.toString()).toEqual('Buffer data')