Allow overriding retryOpts.

[kolodny]

This PR allows overriding retryOpts, makes use of retryOpts.currentRetryAttempt and sends request events for each streaming request made.

These changes are required for smart retries on the bigtable client

The changes there will look something as follows:

diff --git a/node_modules/@google-cloud/bigtable/src/table.js b/node_modules/@google-cloud/bigtable/src/table.js
index 2456464..19d249e 100644
--- a/node_modules/@google-cloud/bigtable/src/table.js
+++ b/node_modules/@google-cloud/bigtable/src/table.js
@@ -28,6 +28,7 @@ var flatten = require('lodash.flatten');
 var is = require('is');
 var propAssign = require('prop-assign');
 var pumpify = require('pumpify');
+var retryRequest = require('retry-request');
 var through = require('through2');
 var util = require('util');

@@ -904,45 +905,74 @@ Table.prototype.insert = function(entries, callback) {
 Table.prototype.mutate = function(entries, callback) {
   entries = flatten(arrify(entries));

-  var grpcOpts = {
-    service: 'Bigtable',
-    method: 'mutateRows'
-  };
-
-  var reqOpts = {
-    objectMode: true,
-    tableName: this.id,
-    entries: entries.map(Mutation.parse)
-  };
-
-  var mutationErrors = [];
-
-  this.requestStream(grpcOpts, reqOpts)
-    .on('error', callback)
-    .on('data', function(obj) {
-      obj.entries.forEach(function(entry) {
-        // Mutation was successful.
-        if (entry.status.code === 0) {
-          return;
-        }
+  var requestsMade = 0

-        var status = commonGrpc.Service.decorateStatus_(entry.status);
-        status.entry = entries[entry.index];
+  var maxRetries = this.maxRetries || 3;
+  var pendingEntryIndices = new Set(entries.map((entry, index) => index));
+  var entryToIndex = new WeakMap(entries.map((entry, index) => [entry, index]));
+  var mutationErrorsByEntryIndex = {};

-        mutationErrors.push(status);
+  var onBatchResponse = () => {
+    if (pendingEntryIndices.size !== 0 && requestsMade <= maxRetries) {
+      setTimeout(
+        makeNextRequestBatch,
+        retryRequest.getNextRetryDelay(requestsMade)
+      );
+      return;
+    }
+    var err = null;
+    var mutationErrors = Object.keys(mutationErrorsByEntryIndex)
+      .filter(entryIndex => mutationErrorsByEntryIndex[entryIndex])
+      .map(entryIndex => mutationErrorsByEntryIndex[entryIndex]);
+
+    if (mutationErrors.length > 0) {
+      err = new common.util.PartialFailureError({
+        errors: mutationErrors
       });
-    })
-    .on('end', function() {
-      var err = null;
+    }
+    callback(err);

-      if (mutationErrors.length > 0) {
-        err = new common.util.PartialFailureError({
-          errors: mutationErrors
-        });
+  };
+
+  var makeNextRequestBatch = () => {
+    var entryBatch = entries.filter((entry, index) => pendingEntryIndices.has(index));
+    var grpcOpts = {
+      service: 'Bigtable',
+      method: 'mutateRows',
+      entries: entryBatch.map(Mutation.parse),
+      retryOpts: {
+        currentRetryAttempt: requestsMade
       }
+    };
+    var reqOpts = {
+      objectMode: true,
+      tableName: this.id,
+      entries: entryBatch.map(Mutation.parse)
+    };
+    this.requestStream(grpcOpts, reqOpts)
+      .on('request', () => requestsMade++)
+      .on('data', function(obj) {
+        obj.entries.forEach(function(entry) {
+          var originalEntry = entryBatch[entry.index]
+          var originalEntriesIndex = entryToIndex.get(originalEntry);
+
+          // Mutation was successful.
+          if (entry.status.code === 0) {
+            pendingEntryIndices.delete(originalEntriesIndex);
+            mutationErrorsByEntryIndex[originalEntriesIndex] = null;
+            return;
+          }
+
+          var status = commonGrpc.Service.decorateStatus_(entry.status);
+          status.entry = mutationEntry;
+          mutationErrorsByEntryIndex[originalEntriesIndex] = status
+        });
+      })
+      .on('end', onBatchResponse)
+      .on('error', onBatchResponse);
+  };

-      callback(err);
-    });
+  makeNextRequestBatch();
 };

[coveralls]

Coverage remained the same at 100.0% when pulling 52e8bcf on kolodny:resumable-requests into b46e673 on GoogleCloudPlatform:master.

[coveralls]

Coverage remained the same at 100.0% when pulling 93cf705 on kolodny:resumable-requests into b46e673 on GoogleCloudPlatform:master.

[coveralls]

Coverage remained the same at 100.0% when pulling 2cb7431 on kolodny:resumable-requests into b46e673 on GoogleCloudPlatform:master.

[kolodny]

Would it be possible to push a release too? These changes are required for Bigtable smart retries on mutate

[stephenplusplus]

@google-cloud/[email protected] released.