pubsub: api redesign #2380

callmehiphop · 2017-06-12T19:12:54Z

TODO:

Unit Tests
StreamingPull System Tests
Docs
README

lukesneeringer

Good overall; a couple concerns with what is there so far.

packages/pubsub/src/message.js

+ *
+ */
+Message.prototype.nack = function() {
+


packages/pubsub/src/publisher.js

+    }
+  });
+
+  this.settings = extend(true, {


packages/pubsub/src/publisher.js

+ */
+Publisher.prototype.publish = function(data, attrs, callback) {
+  if (!(data instanceof Buffer)) {
+    throw new Error('Data must be in the form of a Buffer.');


packages/pubsub/src/publisher.js

+    })
+  };
+
+  this.inventory.inFlight = messages;


packages/pubsub/src/batch.js

@@ -0,0 +1,220 @@
+/*!
+ * Copyright 2014 Google Inc. All Rights Reserved.


packages/pubsub/src/batch.js

+var prop = require('propprop');
+
+/**
+ * Semi-generic Batch object used to queue up multiple requests while limiting


packages/pubsub/src/batch.js

+ *     request.
+ * @param {number} options.maxMilliseconds - Max time to wait before sending
+ *     data.
+ * @param {number} options.maxRequests - Max number of concurrent requests


packages/pubsub/src/batch.js

+ * var messages = batch.getNextMessageBatch();
+ * batch.send(messages);
+ */
+Batch.prototype.getNextMessageBatch = function() {


packages/pubsub/src/batch.js

+    callback: callback
+  });
+
+  var reachedMaxMessages = this.inventory.queued.size >= this.maxMessages;


packages/pubsub/src/batch.js

+  this.intervalHandle_ = setInterval(function() {
+    self.send(self.getNextMessageBatch());
+
+    if (self.inventory.queued.size === 0) {


packages/pubsub/src/batch.js

+  var message;
+
+  while (message = queued.next().value) {
+    var isOverSizeLimit = (size + message.size) > this.maxBytes;


packages/pubsub/src/batch.js

+ * @example
+ * batch.beginSending();
+ */
+Batch.prototype.beginSending = function() {


packages/pubsub/src/batch.js

+Batch.prototype.send = function(messages) {
+  if (this.activeRequests_ >= this.maxRequests) {
+    this.scheduleSend();
+    return;


packages/pubsub/src/message.js

+    ackDeadlineSeconds: seconds
+  };
+
+  this.subscription.request({


packages/pubsub/src/message.js

@@ -0,0 +1,98 @@
+/*!
+ * Copyright 2014 Google Inc. All Rights Reserved.


packages/pubsub/src/publisher.js

@@ -0,0 +1,110 @@
+/*!
+ * Copyright 2014 Google Inc. All Rights Reserved.


packages/pubsub/src/publisher.js

+  this.api = topic.api;
+
+  var batchOptions = extend(options.batching, {
+    send: this.publish_.bind(this)


packages/pubsub/src/subscription.js

-    // times out around 90 seconds. Allow an extra couple of seconds to give the
-    // API a chance to respond on its own before terminating the connection.
-    this.timeout = PUBSUB_API_TIMEOUT + 2000;
+  if (options.topic) {


packages/pubsub/src/index.js

-  commonGrpc.Service.call(this, config, options);
+  this.api = {};
+  this.auth = googleAuth(this.options);
+  this.projectId = this.options.projectId || '{{projectId}}';


packages/pubsub/src/index.js

+ *     used when converting a message's data to a string. (default: 'utf-8')
+ * @param {number|date} options.messageRetentionDuration - Set this to override
+ *     the default duration of 7 days. This value is expected in seconds.
+ *     Acceptable values are in the range of 10 minutes and 7 days.


packages/pubsub/src/publisher.js

+/**
+ * This should never be called directly.
+ */
+Publisher.prototype.publish_ = function(messages, done) {


packages/pubsub/src/subscription.js

+    gaxOpts: gaxOpts
+  }, function(err, resp) {
+    if (!err) {
+      self.closeConnection_();


packages/pubsub/src/subscription.js

    if (err) {
-      callback(err, resp);
+      self.emit('error', err);
      return;


packages/pubsub/src/subscription.js

+      self.emit('error', err);
+    });
+
+    self.connection.on('data', function(data) {


packages/pubsub/src/subscription.js

@@ -678,63 +560,40 @@ Subscription.prototype.delete = function(callback) {
 */
 Subscription.prototype.pull = function(options, callback) {


packages/pubsub/src/subscription.js


-    callback(null, resp);
+    self.connection.on('error', function(err) {
+      var retryCodes = [2, 4, 14];


packages/pubsub/src/subscription.js

-
-    setTimeout(self.startPulling_.bind(self), self.interval);
-  });
+  return this.pubsub.snapshot.call(this, name);
 };



jganetsk · 2017-06-23T15:40:54Z

fyi, gRPC apparently has a default receiver limit on message size of 4 MB, but we need to set the limit to 20 MB + 1 byte.

coveralls · 2017-08-24T17:44:23Z

Coverage remained the same at 100.0% when pulling f758d07 on callmehiphop:dg--pubsub-redesign into f243e26 on GoogleCloudPlatform:master.

callmehiphop added api: pubsub Issues related to the Pub/Sub API. status: do not merge labels Jun 12, 2017

callmehiphop requested review from lukesneeringer and stephenplusplus June 12, 2017 19:12

googlebot added the cla: yes This human has signed the Contributor License Agreement. label Jun 12, 2017

stephenplusplus mentioned this pull request Jun 12, 2017

Experiencing Extreme (10s+) Latencies with Pub/Sub #1974

Closed

lukesneeringer suggested changes Jun 13, 2017

View reviewed changes