pubsub: borrow request config from GAX

packages/pubsub/src/index.js

@@ -45,6 +45,16 @@ var Subscription = require('./subscription.js');
  */
 var Topic = require('./topic.js');
 
+/**
+ * @type {object} - GAX's default configuration.
+ */
+var GAX_CONFIG = {
+  Publisher: require('./v1/publisher_client_config.json').
+    interfaces['google.pubsub.v1.Publisher'],
+  Subscriber: require('./v1/subscriber_client_config.json').
+    interfaces['google.pubsub.v1.Subscriber']
+};
+
 /**
  * [Cloud Pub/Sub](https://developers.google.com/pubsub/overview) is a
  * reliable, many-to-many, asynchronous messaging service from Cloud
@@ -751,6 +761,24 @@ PubSub.prototype.topic = function(name) {
   return new Topic(this, name);
 };
 
+/**
+ * Intercept the call to {module:common/grpc-service#request}, making sure the
+ * correct timeouts are set.
+ *
+ * @private
+ */
+PubSub.prototype.request = function(protoOpts) {
+  var method = protoOpts.method;
+  var camelCaseMethod = method[0].toUpperCase() + method.substr(1);
+  var config = GAX_CONFIG[protoOpts.service].methods[camelCaseMethod];
+
+  if (is.undefined(arguments[0].timeout)) {
+    arguments[0].timeout = config.timeout_millis;
+  }
+
+  commonGrpc.Service.prototype.request.apply(this, arguments);
+};
+
 /**
  * Determine the appropriate endpoint to use for API requests, first trying the
  * local Pub/Sub emulator environment variable (PUBSUB_EMULATOR_HOST), otherwise
@@ -775,7 +803,7 @@ PubSub.prototype.determineBaseUrl_ = function() {
 
 /*! Developer Documentation
  *
- * These methods can be auto-paginated.
+ * These methods can be agto-paginated.
  */
 common.paginator.extend(PubSub, [
   'getSnapshots',
@@ -789,7 +817,12 @@ common.paginator.extend(PubSub, [
  * that a callback is omitted.
  */
 common.util.promisifyAll(PubSub, {
-  exclude: ['snapshot', 'subscription', 'topic']
+  exclude: [
+    'request',
+    'snapshot',
+    'subscription',
+    'topic'
+  ]
 });
 
 PubSub.Subscription = Subscription;

packages/pubsub/src/subscription.js

@@ -465,7 +465,7 @@ Subscription.prototype.ack = function(ackIds, options, callback) {
     ackIds: ackIds
   };
 
-  this.request(protoOpts, reqOpts, function(err, resp) {
+  this.parent.request(protoOpts, reqOpts, function(err, resp) {
     if (!err) {
       ackIds.forEach(function(ackId) {
         delete self.inProgressAckIds[ackId];
@@ -523,7 +523,7 @@ Subscription.prototype.createSnapshot = function(name, callback) {
     subscription: this.name
   };
 
-  this.request(protoOpts, reqOpts, function(err, resp) {
+  this.parent.request(protoOpts, reqOpts, function(err, resp) {
     if (err) {
       callback(err, null, resp);
       return;
@@ -601,7 +601,7 @@ Subscription.prototype.delete = function(callback) {
     subscription: this.name
   };
 
-  this.request(protoOpts, reqOpts, function(err, resp) {
+  this.parent.request(protoOpts, reqOpts, function(err, resp) {
     if (err) {
       callback(err, resp);
       return;
@@ -696,9 +696,11 @@ Subscription.prototype.pull = function(options, callback) {
     maxMessages: options.maxResults
   };
 
-  this.activeRequest_ = this.request(protoOpts, reqOpts, function(err, resp) {
+  this.activeRequest_ = this.parent.request(protoOpts, reqOpts, function(err) {
     self.activeRequest_ = null;
 
+    var resp = arguments[1];
+
     if (err) {
       if (err.code === 504) {
         // Simulate a server timeout where no messages were received.
@@ -779,7 +781,7 @@ Subscription.prototype.seek = function(snapshot, callback) {
     throw new Error('Either a snapshot name or Date is needed to seek to.');
   }
 
-  this.request(protoOpts, reqOpts, callback);
+  this.parent.request(protoOpts, reqOpts, callback);
 };
 
 /**
@@ -825,7 +827,7 @@ Subscription.prototype.setAckDeadline = function(options, callback) {
     ackDeadlineSeconds: options.seconds
   };
 
-  this.request(protoOpts, reqOpts, function(err, resp) {
+  this.parent.request(protoOpts, reqOpts, function(err, resp) {
     callback(err, resp);
   });
 };

packages/pubsub/src/topic.js

@@ -49,7 +49,6 @@ var IAM = require('./iam.js');
  */
 function Topic(pubsub, name) {
   this.name = Topic.formatName_(pubsub.projectId, name);
-  this.pubsub = pubsub;
 
   var methods = {
     /**
@@ -319,7 +318,7 @@ Topic.prototype.getSubscriptions = function(options, callback) {
   options = options || {};
   options.topic = this;
 
-  return this.pubsub.getSubscriptions(options, callback);
+  return this.parent.getSubscriptions(options, callback);
 };
 
 /**
@@ -353,7 +352,7 @@ Topic.prototype.getSubscriptionsStream = function(options) {
   options = options || {};
   options.topic = this;
 
-  return this.pubsub.getSubscriptionsStream(options);
+  return this.parent.getSubscriptionsStream(options);
 };
 
 /**
@@ -475,7 +474,7 @@ Topic.prototype.publish = function(messages, options, callback) {
       .map(Topic.formatMessage_)
   };
 
-  this.request(protoOpts, reqOpts, function(err, result) {
+  this.parent.request(protoOpts, reqOpts, function(err, result) {
     if (err) {
       callback(err, null, result);
       return;
@@ -541,7 +540,7 @@ Topic.prototype.publish = function(messages, options, callback) {
  * });
  */
 Topic.prototype.subscribe = function(subName, options, callback) {
-  this.pubsub.subscribe(this, subName, options, callback);
+  this.parent.subscribe(this, subName, options, callback);
 };
 
 /**
@@ -574,7 +573,7 @@ Topic.prototype.subscription = function(name, options) {
   options = options || {};
   options.topic = this;
 
-  return this.pubsub.subscription(name, options);
+  return this.parent.subscription(name, options);
 };
 
 /*! Developer Documentation

packages/pubsub/test/index.js

@@ -40,14 +40,24 @@ var fakeUtil = extend({}, util, {
     }
 
     promisified = true;
-    assert.deepEqual(options.exclude, ['snapshot', 'subscription', 'topic']);
+    assert.deepEqual(options.exclude, [
+      'request',
+      'snapshot',
+      'subscription',
+      'topic'
+    ]);
   }
 });
 
 function FakeGrpcService() {
   this.calledWith_ = arguments;
 }
 
+var grpcServiceRequestOverride;
+FakeGrpcService.prototype.request = function() {
+  return (grpcServiceRequestOverride || util.noop).apply(this, arguments);
+};
+
 function FakeSnapshot() {
   this.calledWith_ = arguments;
 }
@@ -74,6 +84,22 @@ var fakePaginator = {
   }
 };
 
+var GAX_CONFIG_PUBLISHER_OVERRIDE = {};
+var GAX_CONFIG_SUBSCRIBER_OVERRIDE = {};
+
+var GAX_CONFIG = {
+  Publisher: {
+    interfaces: {
+      'google.pubsub.v1.Publisher': GAX_CONFIG_PUBLISHER_OVERRIDE
+    }
+  },
+  Subscriber: {
+    interfaces: {
+      'google.pubsub.v1.Subscriber': GAX_CONFIG_SUBSCRIBER_OVERRIDE
+    }
+  }
+};
+
 describe('PubSub', function() {
   var PubSub;
   var PROJECT_ID = 'test-project';
@@ -94,7 +120,10 @@ describe('PubSub', function() {
       },
       './snapshot.js': FakeSnapshot,
       './subscription.js': Subscription,
-      './topic.js': Topic
+      './topic.js': Topic,
+
+      './v1/publisher_client_config.json': GAX_CONFIG.Publisher,
+      './v1/subscriber_client_config.json': GAX_CONFIG.Subscriber
     });
   });
 
@@ -105,6 +134,7 @@ describe('PubSub', function() {
   });
 
   beforeEach(function() {
+    grpcServiceRequestOverride = null;
     SubscriptionOverride = null;
     pubsub = new PubSub(OPTIONS);
     pubsub.projectId = PROJECT_ID;
@@ -897,6 +927,86 @@ describe('PubSub', function() {
     });
   });
 
+  describe('request', function() {
+    var TIMEOUT = Math.random();
+
+    beforeEach(function() {
+      GAX_CONFIG_PUBLISHER_OVERRIDE.methods = {
+        MethodName: {
+          timeout_millis: TIMEOUT
+        }
+      };
+    });
+
+    after(function() {
+      GAX_CONFIG_PUBLISHER_OVERRIDE.methods = {};
+    });
+
+    it('should pass through the request', function(done) {
+      var args = [
+        {
+          service: 'Publisher',
+          method: 'MethodName'
+        },
+        {
+          value: true
+        },
+        {
+          anotherValue: true
+        }
+      ];
+
+      grpcServiceRequestOverride = function() {
+        assert.strictEqual(this, pubsub);
+        assert.strictEqual(args[0], arguments[0]);
+        assert.strictEqual(args[1], arguments[1]);
+        assert.strictEqual(args[2], arguments[2]);
+        done();
+      };
+
+      pubsub.request.apply(pubsub, args);
+    });
+
+    it('should assign a timeout', function(done) {
+      grpcServiceRequestOverride = function(protoOpts) {
+        assert.strictEqual(protoOpts.timeout, TIMEOUT);
+        done();
+      };
+
+      pubsub.request({
+        service: 'Publisher',
+        method: 'MethodName'
+      });
+    });
+
+    it('should not override a timeout if set', function(done) {
+      var timeout = 0;
+
+      grpcServiceRequestOverride = function(protoOpts) {
+        assert.strictEqual(protoOpts.timeout, timeout);
+        done();
+      };
+
+      pubsub.request({
+        service: 'Publisher',
+        method: 'MethodName',
+        timeout: timeout
+      });
+    });
+
+    it('should camel case the method name', function(done) {
+      grpcServiceRequestOverride = function(protoOpts) {
+        assert.strictEqual(protoOpts.timeout, TIMEOUT);
+        done();
+      };
+
+      pubsub.request({
+        service: 'Publisher',
+        method: 'methodName'
+      });
+    });
+  });
+
   describe('determineBaseUrl_', function() {
     function setHost(host) {
       process.env.PUBSUB_EMULATOR_HOST = host;