messages are now passed to pull callback & listen for events

Author: stephenplusplus

lib/pubsub/subscription.js

@@ -93,8 +93,6 @@ var util = require('../common/util.js');
  * //-
  * // Once you have obtained a subscription object, you may begin to register
  * // listeners. This will automatically trigger pulling for messages.
- * //
- * // This is a new paragraph.
  * //-
  *
  * // Register an error handler.
@@ -117,7 +115,7 @@ function Subscription(pubsub, options) {
   this.closed = false;
   this.interval = util.is(options.interval, 'number') ? options.interval : 10;
 
-  this.once('newListener', this.startPulling_.bind(this));
+  this.listenForEvents_();
 }
 
 nodeutil.inherits(Subscription, events.EventEmitter);
@@ -136,6 +134,95 @@ Subscription.formatName_ = function(projectId, name) {
   return '/subscriptions/' + projectId + '/' + name;
 };
 
+/**
+ * Simplify a message from an API response to have two properties, `id` and
+ * `data`. `data` is always converted to a string.
+ *
+ * @private
+ */
+Subscription.formatMessage_ = function(msg) {
+  var message = {
+    id: msg.ackId
+  };
+  if (msg.pubsubEvent && msg.pubsubEvent.message) {
+    message.data =
+      new Buffer(msg.pubsubEvent.message.data, 'base64').toString('utf-8');
+    try {
+      message.data = JSON.parse(message.data);
+    } catch(e) {}
+  }
+  return message;
+};
+
+/**
+ * Begin listening for events on the subscription. This method keeps track of
+ * how many message listeners are assigned, and then removed, making sure
+ * polling is handled automatically.
+ *
+ * As long as there is one active message listener, the connection is open. As
+ * soon as there are no more message listeners, the connection is closed.
+ *
+ * @private
+ *
+ * @example
+ * this.listenForEvents_();
+ */
+Subscription.prototype.listenForEvents_ = function() {
+  var that = this;
+  var messageListeners = 0;
+
+  this.on('newListener', function(event) {
+    if (event === 'message') {
+      messageListeners++;
+      if (that.closed) {
+        that.closed = false;
+      }
+      that.startPulling_();
+    }
+  });
+
+  this.on('removeListener', function(event) {
+    if (event === 'message' && --messageListeners === 0) {
+      that.closed = true;
+    }
+  });
+};
+
+/**
+ * Poll the backend for new messages. This runs a loop to ping the API at the
+ * provided interval from the subscription's instantiation. If one wasn't
+ * provided, the default value is 10 milliseconds.
+ *
+ * If messages are received, they are emitted on the `message` event.
+ *
+ * Note: This method is automatically called once a message event handler is
+ * assigned to the description.
+ *
+ * To stop pulling, see {@linkcode module:pubsub/subscription#close}.
+ *
+ * @private
+ *
+ * @example
+ * subscription.startPulling_();
+ */
+Subscription.prototype.startPulling_ = function() {
+  var that = this;
+  if (this.closed) {
+    return;
+  }
+  this.pull({
+    returnImmediately: false
+  }, function(err, message) {
+    if (err) {
+      that.emit('error', err);
+    }
+    if (message) {
+      that.emit('message', message);
+    }
+    setTimeout(that.startPulling_.bind(that), that.interval);
+  });
+};
+
 /**
  * Acknowledge to the backend that the message was retrieved. You must provide
  * either a single ID, or an array of IDs.
@@ -162,15 +249,35 @@ Subscription.prototype.ack = function(ids, callback) {
 };
 
 /**
- * Pull messages from the subscribed topic. If messages were found, they are
- * passed along with a `message` event.
+ * Delete the subscription. Pull requests from the current subscription will be
+ * errored once unsubscription is complete.
+ *
+ * @param {function=} callback - The callback function.
+ *
+ * @example
+ * subscription.delete(function(err) {});
+ */
+Subscription.prototype.delete = function(callback) {
+  callback = callback || util.noop;
+  this.request(
+      'DELETE', 'subscriptions/' + this.name, null, true, function(err) {
+    if (err) {
+      callback(err);
+      return;
+    }
+    this.closed = true;
+    callback(null);
+  }.bind(this));
+};
+
+/**
+ * Pull messages from the subscribed topic. If messages were found, your
+ * callback is executed with the message object.
  *
  * Note that messages are pulled automatically once you register your first
  * event listener to the subscription, thus the call to `pull` is handled for
- * you.
- *
- * Calling `pull` directly can be helpful after your subscription has been
- * closed with {@linkcode module:pubsub/subscription#close}.
+ * you. If you don't want to start pulling, simply don't register a
+ * `subscription.on('message', function() {})` event handler.
  *
  * @param {object=} options - Configuration object.
  * @param {boolean=} options.returnImmediately - If set, the system will respond
@@ -179,7 +286,10 @@ Subscription.prototype.ack = function(ids, callback) {
  * @param {function} callback - The callback function.
  *
  * @example
- * subscription.pull(function(err) {});
+ * subscription.pull(function(err, message) {
+ *   // message.id = ID used to acknowledge its receival.
+ *   // message.data = Contents of the message.
+ * });
  */
 Subscription.prototype.pull = function(options, callback) {
   var that = this;
@@ -199,104 +309,15 @@ Subscription.prototype.pull = function(options, callback) {
       callback(err);
       return;
     }
-    if (!that.autoAck) {
-      that.emitMessage_(message);
-      callback();
-      return;
+    message = Subscription.formatMessage_(message);
+    if (that.autoAck) {
+      that.ack(message.id, function(err) {
+        callback(err, message);
+      });
+    } else {
+      callback(null, message);
     }
-    that.ack(message.ackId, function(err) {
-      if (err) {
-        callback(err);
-        return;
-      }
-      that.emitMessage_(message);
-      callback();
-    });
   });
 };
 
-/**
- * Delete the subscription. Pull requests from the current subscription will be
- * errored once unsubscription is complete.
- *
- * @param {function=} callback - The callback function.
- *
- * @example
- * subscription.delete(function(err) {});
- */
-Subscription.prototype.delete = function(callback) {
-  callback = callback || util.noop;
-  this.request(
-      'DELETE', 'subscriptions/' + this.name, null, true, function(err) {
-    if (err) {
-      callback(err);
-      return;
-    }
-    this.closed = true;
-    callback(null);
-  }.bind(this));
-};
-
-/**
- * Poll the backend for new messages. This runs a loop to ping the API at the
- * provided interval from the subscription's instantiation. If you didn't
- * provide one, the default value is 10 milliseconds.
- *
- * If messages are received, you can catch them by registering a listener for
- * the `message` event.
- *
- * To stop pulling, see {@linkcode module:pubsub/subscription#stopPulling}.
- *
- * @private
- *
- * @example
- * subscription.startPulling_();
- */
-Subscription.prototype.startPulling_ = function() {
-  if (this.closed) {
-    return;
-  }
-  this.pull({
-    returnImmediately: false
-  }, function(err) {
-    if (err && err.code === 400) {
-      this.emit('error', err);
-    }
-    setTimeout(this.startPulling_.bind(this), this.interval);
-  }.bind(this));
-};
-
-/**
- * Stop the subscription from automatically pulling. You will still be able to
- * call {@linkcode module:pubsub/subscription#pull} directly.
- *
- * @example
- * subscription.close();
- */
-Subscription.prototype.close = function() {
-  this.closed = true;
-};
-
-/**
- * Emits a 'message' event with the provided message.
- *
- * The message is simplified from the API response to have simply two
- * properties, `id` and `data`. `data` is always converted to a string.
- *
- * @private
- */
-Subscription.prototype.emitMessage_ = function(msg) {
-  var message = {
-    id: msg.ackId
-  };
-  if (msg.pubsubEvent && msg.pubsubEvent.message) {
-    message.data =
-      new Buffer(msg.pubsubEvent.message.data, 'base64').toString('utf-8');
-    try {
-      message.data = JSON.parse(message.data);
-    } catch(e) {}
-  }
-  this.emit('message', message);
-};
-
 module.exports = Subscription;

regression/pubsub.js

@@ -174,14 +174,16 @@ describe('pubsub', function() {
 
     it('should be able to pull and ack', function(done) {
       var subscription = topic.subscription(subscriptions[0].name);
-      subscription
-        .once('message', function(msg) {
-          subscription.ack(msg.id, done);
-        })
-        .pull({ returnImmediately: true }, assert.ifError);
-      [1, 2, 3, 4, 5, 6].forEach(function() {
-        topic.publish('hello', assert.ifError);
+      subscription.pull({ returnImmediately: true }, function(err, msg) {
+        assert.ifError(err);
+        subscription.ack(msg.id, done);
       });
+      topic.publish('hello', assert.ifError);
+      topic.publish('hello', assert.ifError);
+      topic.publish('hello', assert.ifError);
+      topic.publish('hello', assert.ifError);
+      topic.publish('hello', assert.ifError);
+      topic.publish('hello', assert.ifError);
     });
   });
 });