gcloud subscription

subscribe.js

'use strict';

var topicName = 'book-process-queue';
var subscriptionName = 'shared-worker-subscription';

var creatingSubscription = false;
var subscription;
var onMessageHandlers = [];

module.exports = function() {
  var pubsub = gcloud.pubsub(config.gcloud);
  var topic = pubsub.topic(topicName);

  function subscribe(onMessageHandler) {
    onMessageHandlers.push(onMessageHandler);

    if (!subscription) {
      createSubscription();
    }
  }

  function createSubscription() {
    if (creatingSubscription) {
      return;
    }

    creatingSubscription = true;
    topic.subscribe(subscriptionName, { autoAck: true }, onSubscriptionCreated);
  }

  function onSubscriptionCreated(err, subscriptionInstance) {
    creatingSubscription = false;

    if (err && err.code === 404) {
      pubsub.createTopic(topicName, createSubscription);
      return;
    }

    subscription = subscriptionInstance;

    subscription.on('message', onMessageReceived);

    logging.info('Listening to ' + topicName + ' with subscription ' + subscriptionName);
  }

  function onMessageReceived(message) {
    onMessageHandlers.forEach(function(onMessageHandler) {
      onMessageHandler(message.data);
    });
  }

  return {
    subscribe: subscribe
  };
};