Stream support for FileUpload

lib/Error.js

@@ -73,3 +73,13 @@ _Error.StripeRateLimitError = StripeError.extend({type: 'StripeRateLimitError'})
 _Error.StripeConnectionError = StripeError.extend({type: 'StripeConnectionError'});
 _Error.StripeSignatureVerificationError = StripeError.extend({type: 'StripeSignatureVerificationError'});
 _Error.StripeIdempotencyError = StripeError.extend({type: 'StripeIdempotencyError'});
+
+// Error while processing stream for file upload:
+_Error.StreamProcessingError = _Error.extend({
+  type: 'StreamProcessingError',
+  populate: function(raw) {
+    this.type = this.type;
+    this.message = raw.message;
+    this.detail = raw.detail;
+  }
+});

lib/StripeResource.js

@@ -212,6 +212,20 @@ StripeResource.prototype = {
     }
   },
 
+  _errorHandlerStream: function(callback) {
+    var self = this;
+    return function(error) {
+      callback.call(
+        self,
+        new Error.StreamProcessingError({
+          message: 'An error occurred while attempting to process the file for upload.',
+          detail: error,
+        }),
+        null
+      );
+    }
+  },
+
   _defaultHeaders: function(auth, contentLength, apiVersion) {
     var userAgentString = 'Stripe/v1 NodeBindings/' + this._stripe.getConstant('PACKAGE_VERSION');
 
@@ -241,28 +255,37 @@ StripeResource.prototype = {
     var self = this;
     var requestData;
 
-    if (self.requestDataProcessor) {
-      requestData = self.requestDataProcessor(method, data, options.headers);
-    } else {
-      requestData = utils.stringifyRequestData(data || {});
-    }
+    function makeRequestWithData(error, data) {
+      var apiVersion;
+      var headers;
 
-    var apiVersion = this._stripe.getApiField('version');
+      if (error) {
+        var handleError = self._errorHandlerStream(callback);
+        return handleError(error);
+      }
 
-    var headers = self._defaultHeaders(auth, requestData.length, apiVersion);
+      apiVersion = self._stripe.getApiField('version');
+      requestData = data;
+      headers = self._defaultHeaders(auth, requestData.length, apiVersion);
 
-    // Grab client-user-agent before making the request:
-    this._stripe.getClientUserAgent(function(cua) {
-      headers['X-Stripe-Client-User-Agent'] = cua;
+      self._stripe.getClientUserAgent(function(cua) {
+        headers['X-Stripe-Client-User-Agent'] = cua;
 
-      if (options.headers) {
-        Object.assign(headers, options.headers);
-      }
+        if (options.headers) {
+          Object.assign(headers, options.headers);
+        }
 
-      makeRequest();
-    });
+        makeRequest(apiVersion, headers);
+      });
+    }
+
+    if (self.requestDataProcessor) {
+      self.requestDataProcessor(method, data, options.headers, makeRequestWithData);
+    } else {
+      makeRequestWithData(null, utils.stringifyRequestData(data || {}));
+    }
 
-    function makeRequest() {
+    function makeRequest(apiVersion, headers) {
       var timeout = self._stripe.getApiField('timeout');
       var isInsecureConnection = self._stripe.getApiField('protocol') == 'http';
 

lib/resources/FileUploads.js

@@ -1,5 +1,6 @@
 'use strict';
 
+var Buffer = require('safe-buffer').Buffer;
 var utils = require('../utils');
 var StripeResource = require('../StripeResource');
 var stripeMethod = StripeResource.method;
@@ -9,14 +10,38 @@ module.exports = StripeResource.extend({
 
   overrideHost: 'uploads.stripe.com',
 
-  requestDataProcessor: function(method, data, headers) {
+  requestDataProcessor: function(method, data, headers, callback) {
     data = data || {};
 
     if (method === 'POST') {
-      return multipartDataGenerator(method, data, headers);
+      return getProcessorForSourceType(data);
     } else {
       return utils.stringifyRequestData(data);
     }
+
+    function getProcessorForSourceType(data) {
+      var isStream = utils.checkForStream(data);
+      if (isStream) {
+        return streamProcessor(multipartDataGenerator);
+      } else {
+        var buffer = multipartDataGenerator(method, data, headers);
+        return callback(null, buffer);
+      }
+    }
+
+    function streamProcessor (fn) {
+      var bufferArray = [];
+      data.file.data.on('data', function(line) {
+        bufferArray.push(line);
+      }).on('end', function() {
+        var bufferData = Object.assign({}, data);
+        bufferData.file.data = Buffer.concat(bufferArray);
+        var buffer = fn(method, bufferData, headers);
+        callback(null, buffer);
+      }).on('error', function(err) {
+        callback(err, null);
+      });
+    }
   },
 
   path: 'files',

lib/utils.js

@@ -1,6 +1,7 @@
 'use strict';
 
 var Buffer = require('safe-buffer').Buffer;
+var EventEmitter = require('events').EventEmitter;
 var qs = require('qs');
 var crypto = require('crypto');
 
@@ -215,6 +216,17 @@ var utils = module.exports = {
 
     return obj;
   },
+
+  /**
+  * Determine if file data is a derivative of EventEmitter class.
+  * https://nodejs.org/api/events.html#events_events
+  */
+  checkForStream: function (obj) {
+    if (obj.file && obj.file.data) {
+      return obj.file.data instanceof EventEmitter;
+    }
+    return false;
+  },
 };
 
 function emitWarning(warning) {

test/flows.spec.js

@@ -6,7 +6,8 @@ var stripe = require('../lib/stripe')(
   testUtils.getUserStripeKey(),
   'latest'
 );
-
+var fs = require('fs');
+var path = require('path');
 var expect = chai.expect;
 
 var CUSTOMER_DETAILS = {
@@ -522,4 +523,24 @@ describe('Flows', function() {
       }).type).to.equal('StripeInvalidRequestError');
     });
   });
+
+  describe('FileUpload', function() {
+    it('Allows you to upload a file as a stream', function() {
+      var testFilename = path.join(__dirname, 'resources/data/minimal.pdf');
+      var f = fs.createReadStream(testFilename);
+
+      return expect(
+        stripe.fileUploads.create({
+          purpose: 'dispute_evidence',
+          file: {
+            data: f,
+            name: 'minimal.pdf',
+            type: 'application/octet-stream',
+          },
+        }).then(null, function(error) {
+          return error;
+        })
+      ).to.eventually.have.nested.property('size', 739);
+    });
+  });
 });

test/resources/FileUploads.spec.js

@@ -78,5 +78,40 @@ describe('File Uploads Resource', function() {
       expect(stripe.LAST_REQUEST).to.deep.property('url', '/v1/files');
       expect(stripe.LAST_REQUEST).to.deep.property('auth', TEST_AUTH_KEY);
     });
+
+    it('Streams a file and sends the correct file upload request', function() {
+      var testFilename = path.join(__dirname, 'data/minimal.pdf');
+      var f = fs.createReadStream(testFilename);
+
+      return stripe.fileUploads.create({
+        purpose: 'dispute_evidence',
+        file: {
+          data: f,
+          name: 'minimal.pdf',
+          type: 'application/octet-stream',
+        },
+      }).then(function() {
+        expect(stripe.LAST_REQUEST).to.deep.property('method', 'POST');
+        expect(stripe.LAST_REQUEST).to.deep.property('url', '/v1/files');
+      });
+    });
+
+    it('Streams a file and sends the correct file upload request [with specified auth]', function() {
+      var testFilename = path.join(__dirname, 'data/minimal.pdf');
+      var f = fs.createReadStream(testFilename);
+
+      return stripe.fileUploads.create({
+        purpose: 'dispute_evidence',
+        file: {
+          data: f,
+          name: 'minimal.pdf',
+          type: 'application/octet-stream',
+        },
+      }, TEST_AUTH_KEY).then(function() {
+        expect(stripe.LAST_REQUEST).to.deep.property('method', 'POST');
+        expect(stripe.LAST_REQUEST).to.deep.property('url', '/v1/files');
+        expect(stripe.LAST_REQUEST).to.deep.property('auth', TEST_AUTH_KEY);
+      });
+    });
   });
 });