core: add retry logic to read streams

Author: stephenplusplus

lib/common/util.js

@@ -29,6 +29,9 @@ var request = require('request').defaults({
     maxSockets: Infinity
   }
 });
+var retryRequest = require('retry-request');
+var streamForward = require('stream-forward');
+var through = require('through2');
 var uuid = require('node-uuid');
 
 /** @const {object} gcloud-node's package.json file. */
@@ -178,34 +181,68 @@ nodeutil.inherits(ApiError, Error);
  */
 function handleResp(err, resp, body, callback) {
   callback = callback || noop;
+
   if (err) {
     callback(err);
     return;
   }
-  if (typeof body === 'string') {
-    try {
-      body = JSON.parse(body);
-    } catch(err) {}
-  }
-  if (body && body.error) {
-    // Error from JSON api.
-    callback(new ApiError(body.error));
+
+  var parsedApiResponse = util.parseApiResp(resp, body);
+
+  if (parsedApiResponse.err) {
+    callback(parsedApiResponse.err);
     return;
   }
-  if (resp && (resp.statusCode < 200 || resp.statusCode > 299)) {
+
+  callback(null, parsedApiResponse.body, parsedApiResponse.resp);
+}
+
+util.handleResp = handleResp;
+
+/**
+ * From an HTTP response, generate an error if one occurred.
+ *
+ * @param {*} resp - Response value.
+ * @param {*=} body - Body value.
+ * @return {object} parsedResponse - The parsed response.
+ * @param {?error} parsedResponse.err - An error detected.
+ * @param {object} parsedResponse.resp - The original response object.
+ * @param {*} parsedREsponse.body - The original body value provided will try to
+ *     be JSON.parse'd. If it's successful, the parsed value will be returned
+ *     here, otherwise the original value.
+ */
+function parseApiResp(resp, body) {
+  var parsedResponse = {
+    err: null,
+    resp: resp,
+    body: body
+  };
+
+  if (resp.statusCode < 200 || resp.statusCode > 299) {
     // Unknown error. Format according to ApiError standard.
-    callback(new ApiError({
+    parsedResponse.err = new ApiError({
       errors: [],
       code: resp.statusCode,
-      message: body || 'Error during request.',
+      message: 'Error during request.',
       response: resp
-    }));
-    return;
+    });
+  }
+
+  if (util.is(body, 'string')) {
+    try {
+      parsedResponse.body = JSON.parse(body);
+    } catch(err) {}
+  }
+
+  if (parsedResponse.body && parsedResponse.body.error) {
+    // Error from JSON API.
+    parsedResponse.err = new ApiError(parsedResponse.body.error);
   }
-  callback(null, body, resp);
+
+  return parsedResponse;
 }
 
-util.handleResp = handleResp;
+util.parseApiResp = parseApiResp;
 
 /**
  * Get the type of a value.
@@ -418,20 +455,6 @@ function makeWritableStream(dup, options, onComplete) {
 
 util.makeWritableStream = makeWritableStream;
 
-/**
- * Returns an exponential distributed time to wait given the number of retries
- * that have been previously been attempted on the request.
- *
- * @param {number} retryNumber - The number of retries previously attempted.
- * @return {number} An exponentially distributed time to wait E.g. for use with
- *     exponential backoff.
- */
-function getNextRetryWait(retryNumber) {
-  return (Math.pow(2, retryNumber) * 1000) + Math.floor(Math.random() * 1000);
-}
-
-util.getNextRetryWait = getNextRetryWait;
-
 /**
  * Returns true if the API request should be retried, given the error that was
  * given the first time the request was attempted. This is used for rate limit
@@ -599,31 +622,45 @@ function makeAuthorizedRequestFactory(config) {
    *     be made. Instead, this function is passed the error & authorized
    *     request options.
    */
-  function makeAuthorizedRequest(reqOpts, callback) {
-    if (config.customEndpoint) {
-      // Using a custom API override. Do not use `google-auth-library` for
-      // authentication. (ex: connecting to a local Datastore server)
-      if (callback.onAuthorized) {
-        callback.onAuthorized(null, reqOpts);
-      } else {
-        util.makeRequest(reqOpts, config, callback);
-      }
+  function makeAuthorizedRequest(reqOpts, options) {
+    var stream;
+    var reqConfig = extend({}, config);
 
-      return;
+    if (!options) {
+      stream = through();
+      reqConfig.stream = stream;
     }
 
-    util.authorizeRequest(config, reqOpts, function(err, authorizedReqOpts) {
+    function onAuthorized(err, authorizedReqOpts) {
       if (err) {
-        (callback.onAuthorized || callback)(err);
+        if (stream) {
+          stream.emit('error', err);
+          stream.end();
+        } else {
+          (options || options.onAuthorized)(err);
+        }
+
         return;
       }
 
-      if (callback.onAuthorized) {
-        callback.onAuthorized(null, authorizedReqOpts);
+      if (options && options.onAuthorized) {
+        options.onAuthorized(null, authorizedReqOpts);
       } else {
-        util.makeRequest(authorizedReqOpts, config, callback);
+        util.makeRequest(authorizedReqOpts, reqConfig, options);
       }
-    });
+    }
+
+    if (reqConfig.customEndpoint) {
+      // Using a custom API override. Do not use `google-auth-library` for
+      // authentication. (ex: connecting to a local Datastore server)
+      onAuthorized(null, reqOpts);
+    } else {
+      util.authorizeRequest(reqConfig, reqOpts, onAuthorized);
+    }
+
+    if (stream) {
+      return stream;
+    }
   }
 
   makeAuthorizedRequest.getCredentials = function(callback) {
@@ -653,8 +690,8 @@ function makeAuthorizedRequestFactory(config) {
 util.makeAuthorizedRequestFactory = makeAuthorizedRequestFactory;
 
 /**
- * Make a request through the `request` module with built-in error handling and
- * exponential back off.
+ * Make a request through the `retryRequest` module with built-in error handling
+ * and exponential back off.
  *
  * @param {object} reqOpts - Request options in the format `request` expects.
  * @param {object=} config - Configuration object.
@@ -676,30 +713,26 @@ function makeRequest(reqOpts, config, callback) {
 
   reqOpts = util.decorateRequest(reqOpts);
 
-  var MAX_RETRIES = config.maxRetries || 3;
-  var autoRetry = config.autoRetry !== false ? true : false;
-  var attemptedRetries = 0;
+  var options = {
+    request: request,
 
-  function shouldRetry(err) {
-    return autoRetry &&
-      MAX_RETRIES > attemptedRetries &&
-      util.shouldRetryRequest(err);
-  }
+    retries: config.autoRetry !== false ? config.maxRetries || 3 : 0,
 
-  function makeRateLimitedRequest() {
-    request(reqOpts, function(err, resp, body) {
-      util.handleResp(err, resp, body, function(err, body, resp) {
-        if (shouldRetry(err)) {
-          var delay = util.getNextRetryWait(attemptedRetries++);
-          setTimeout(makeRateLimitedRequest, delay);
-        } else {
-          callback(err || null, body, resp);
-        }
-      });
+    shouldRetryFn: function(resp) {
+      var err = util.parseApiResp(resp).err;
+      return err && util.shouldRetryRequest(err);
+    }
+  };
+
+  if (config.stream) {
+    // `streamForward` is used to re-emit the events the request stream receives
+    // on to the stream the user is holding (config.stream).
+    streamForward(retryRequest(reqOpts, options)).pipe(config.stream);
+  } else {
+    retryRequest(reqOpts, options, function(err, response, body) {
+      util.handleResp(err, response, body, callback);
     });
   }
-
-  makeRateLimitedRequest();
 }
 
 util.makeRequest = makeRequest;