Allow overriding retryOpts. (#2764)

* Allow overriding `retryOpts`.

* Removed extra newline

* Use retry-request request events

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.3.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,8 +373,9 @@ GrpcService.prototype.requestStream = function(protoOpts, reqOpts) {
     return stream;
   }
 
-  var retryOpts = {
+  var retryOpts = extend({
     retries: this.maxRetries,
+    currentRetryAttempt: 0,
     objectMode: objectMode,
     shouldRetryFn: GrpcService.shouldRetryRequest_,
 
@@ -391,14 +393,15 @@ GrpcService.prototype.requestStream = function(protoOpts, reqOpts) {
           this.emit('response', grcpStatus);
         });
     }
-  };
+  }, protoOpts.retryOpts);
 
   return retryRequest(null, retryOpts)
     .on('error', function(err) {
       var grpcError = GrpcService.decorateError_(err);
 
       stream.destroy(grpcError || err);
     })
+    .on('request', stream.emit.bind(stream, 'request'))
     .pipe(stream);
 };
 

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,16 @@ describe('GrpcService', function() {
         fakeStream.emit('metadata');
       });
 
+      it('should forward `request` events', function(done) {
+        var requestStream = grpcService.requestStream(PROTO_OPTS, REQ_OPTS);
+
+        requestStream.on('request', function() {
+          done()
+        });
+
+        retryStream.emit('request');
+      });
+
       it('should emit the response error', function(done) {
         var grpcError500 = { code: 2 };
         var requestStream = grpcService.requestStream(PROTO_OPTS, REQ_OPTS);