add downloadStream API

andrewrk · andrewrk · commit 372f49196f16 · 2014-11-03T14:15:59.000-07:00
closes andrewrk#53
diff --git a/README.md b/README.md
@@ -308,6 +308,23 @@ And these events:
  * `'progress'` - emitted when `progressAmount` and `progressTotal`
    properties change.
 
+### client.downloadStream(s3Params)
+
+http://docs.aws.amazon.com/AWSJavaScriptSDK/latest/AWS/S3.html#getObject-property
+
+ * `s3Params`: params to pass to AWS SDK `getObject`.
+
+The difference between using AWS SDK `getObject` and this one:
+
+ * This works with a stream only.
+
+If you want retries, progress, or MD5 checking, you must code it yourself.
+
+Returns a `ReadableStream` with these additional events:
+
+ * `'httpHeaders' (statusCode, headers)` - contains the HTTP response
+   headers and status code.
+
 ### client.listObjects(params)
 
 See http://docs.aws.amazon.com/AWSJavaScriptSDK/latest/AWS/S3.html#listObjects-property
diff --git a/lib/index.js b/lib/index.js
@@ -13,6 +13,7 @@ var MultipartETag = require('./multipart_etag');
 var FdSlicer = require('fd-slicer');
 var mime = require('mime');
 var StreamSink = require('streamsink');
+var PassThrough = require('stream').PassThrough;
 
 var MAX_PUTOBJECT_SIZE = 5 * 1024 * 1024 * 1024;
 var MAX_DELETE_COUNT = 1000;
@@ -840,6 +841,61 @@ Client.prototype.downloadBuffer = function(s3Params) {
   }
 };
 
+Client.prototype.downloadStream = function(s3Params) {
+  var self = this;
+  var downloadStream = new PassThrough();
+  s3Params = extend({}, s3Params);
+
+  doDownloadWithPend(function(err) {
+    if (err) downloadStream.emit('error', err);
+  });
+  return downloadStream;
+
+  function doDownloadWithPend(cb) {
+    self.s3Pend.go(function(pendCb) {
+      doTheDownload(function(err) {
+        pendCb();
+        cb(err);
+      });
+    });
+  }
+
+  function doTheDownload(cb) {
+    var errorOccurred = false;
+    var request = self.s3.getObject(s3Params);
+    var hashCheckPend = new Pend();
+    request.on('build', function() {
+      request.httpRequest.headers.Expect = '100-continue';
+    });
+    request.on('httpHeaders', function(statusCode, headers, resp) {
+      if (statusCode >= 300) {
+        handleError(new Error("http status code " + statusCode));
+        return;
+      }
+      downloadStream.emit('httpHeaders', statusCode, headers, resp);
+      var httpStream = resp.httpResponse.createUnbufferedStream();
+
+      httpStream.on('error', handleError);
+
+      downloadStream.on('finish', function() {
+        if (errorOccurred) return;
+        cb();
+      });
+
+      httpStream.pipe(downloadStream);
+    });
+
+    request.send(handleError);
+
+    function handleError(err) {
+      if (!err) return;
+      if (errorOccurred) return;
+      errorOccurred = true;
+      cb(err);
+    }
+  }
+};
+
 function syncDir(self, params, directionIsToS3) {
   var ee = new EventEmitter();
   var finditOpts = {
diff --git a/test/test.js b/test/test.js
@@ -8,6 +8,7 @@ var fs = require('fs');
 var mkdirp = require('mkdirp');
 var crypto = require('crypto');
 var rimraf = require('rimraf');
+var StreamSink = require('streamsink');
 var tempDir = path.join(__dirname, 'tmp');
 var localFile = path.join(tempDir, 'random.png');
 var remoteRoot = "node-s3-test/";
@@ -188,6 +189,27 @@ describe("s3", function () {
     });
   });
 
+  it("downloadStream", function(done) {
+    var client = createClient();
+    var downloadStream = client.downloadStream({Key: remoteFile, Bucket: s3Bucket});
+    downloadStream.on('error', done);
+    var gotHttpHeaders = false;
+    downloadStream.on('httpHeaders', function(statusCode, headers, resp) {
+      var contentType = headers['content-type'];
+      assert.strictEqual(contentType, "image/png");
+      gotHttpHeaders = true;
+    });
+    var sink = new StreamSink();
+    downloadStream.pipe(sink);
+    sink.on('finish', function() {
+      var md5sum = crypto.createHash('md5');
+      md5sum.update(sink.toBuffer());
+      assert.strictEqual(md5sum.digest('hex'), hexdigest)
+      assert.ok(gotHttpHeaders);
+      done();
+    });
+  });
+
   it("lists objects", function(done) {
     var params = {
       recursive: true,
