Allow overriding retryOpts.

Author: kolodny

packages/common-grpc/package.json

@@ -45,7 +45,7 @@
     "grpc": "^1.7.1",
     "is": "^3.2.0",
     "modelo": "^4.2.0",
-    "retry-request": "^3.0.0",
+    "retry-request": "^3.2.0",
     "through2": "^2.0.3"
   },
   "devDependencies": {

packages/common-grpc/src/service.js

@@ -272,8 +272,9 @@ GrpcService.prototype.request = function(protoOpts, reqOpts, callback) {
   // executed with this as the "response", we return it to the user as an error.
   var respError;
 
-  var retryOpts = {
+  var retryOpts = extend({
     retries: this.maxRetries,
+    currentRetryAttempt: 0,
     shouldRetryFn: GrpcService.shouldRetryRequest_,
 
     // retry-request determines if it should retry from the incoming HTTP
@@ -303,7 +304,7 @@ GrpcService.prototype.request = function(protoOpts, reqOpts, callback) {
           onResponse(null, resp);
         });
     }
-  };
+  }, protoOpts.retryOpts);
 
   return retryRequest(null, retryOpts, function(err, resp) {
     if (!err && resp === respError) {
@@ -372,12 +373,14 @@ GrpcService.prototype.requestStream = function(protoOpts, reqOpts) {
     return stream;
   }
 
-  var retryOpts = {
+  var retryOpts = extend({
     retries: this.maxRetries,
+    currentRetryAttempt: 0,
     objectMode: objectMode,
     shouldRetryFn: GrpcService.shouldRetryRequest_,
 
     request: function() {
+      setImmediate(() => stream.emit('request'));
       return service[protoOpts.method](reqOpts, grpcMetadata, grpcOpts)
         .on('metadata', function() {
           // retry-request requires a server response before it starts emitting
@@ -391,7 +394,7 @@ GrpcService.prototype.requestStream = function(protoOpts, reqOpts) {
           this.emit('response', grcpStatus);
         });
     }
-  };
+  }, protoOpts.retryOpts);
 
   return retryRequest(null, retryOpts)
     .on('error', function(err) {

packages/common-grpc/test/service.js

@@ -617,6 +617,7 @@ describe('GrpcService', function() {
           retryRequestOptions.retries,
           grpcService.maxRetries
         );
+        assert.strictEqual(retryRequestOptions.currentRetryAttempt, 0);
 
         retryRequestCallback(error, response);
       });
@@ -1082,6 +1083,7 @@ describe('GrpcService', function() {
           retryRequestOptions.retries,
           grpcService.maxRetries
         );
+        assert.strictEqual(retryRequestOptions.currentRetryAttempt, 0);
         assert.strictEqual(retryRequestOptions.objectMode, true);
         assert.strictEqual(
           retryRequestOptions.shouldRetryFn,
@@ -1111,6 +1113,34 @@ describe('GrpcService', function() {
         fakeStream.emit('metadata');
       });
 
+      it('should emit a `request` event on each request', function(done) {
+        var fakeStream = through.obj();
+
+        ProtoService.prototype.method = function() {
+          return fakeStream;
+        };
+
+        retryRequestOverride = function(reqOpts, options) {
+          // Simulate three retries.
+          setImmediate(function() {
+            options.request();
+            options.request();
+          });
+          return options.request();
+        };
+
+        var requestStream = grpcService.requestStream(PROTO_OPTS, REQ_OPTS);
+
+        var requestEmitted = 0;
+        requestStream.on('request', function() {
+          requestEmitted++;
+          if (requestEmitted === 3) {
+            done();
+          }
+        });
+
+      });
+
       it('should emit the response error', function(done) {
         var grpcError500 = { code: 2 };
         var requestStream = grpcService.requestStream(PROTO_OPTS, REQ_OPTS);