Initial commit.

Author: jmdobry

pubsub/subscriptions.js

@@ -168,6 +168,62 @@ function pullMessages (subscriptionName, callback) {
 }
 // [END pubsub_pull_messages]
 
+let subscribeCounterValue = 1;
+
+function getSubscribeCounterValue () {
+ return subscribeCounterValue;
+}
+
+function setSubscribeCounterValue (value) {
+ subscribeCounterValue = value;
+}
+
+// [START pubsub_pull_ordered_messages]
+var outstandingMessages = {};
+
+function pullOrderedMessages (subscriptionName, callback) {
+ // References an existing subscription, e.g. "my-subscription"
+ const subscription = pubsubClient.subscription(subscriptionName);
+
+ // Pulls messages. Set returnImmediately to false to block until messages are
+ // received.
+ subscription.pull({ returnImmediately: true }, (err, messages) => {
+ if (err) {
+ callback(err);
+ return;
+ }
+
+ // Sort messages in order of increasing messageId
+ messages.sort((a, b) => b.messageId - a.messageId);
+
+ // Iterate over messages in order of increasing messageId
+ messages.forEach((message) => outstandingMessages[message.messageId] = message);
+
+ const outstandingMessageIds = Object.keys(outstandingMessages);
+ outstandingMessageIds.sort();
+
+ outstandingMessageIds.forEach((messageId) => {
+ const counter = getSubscribeCounterValue();
+ const message = outstandingMessages[messageId];
+
+ if (messageId < counter) {
+ // The message has already been processed
+ subscription.ack(message.ackId);
+ delete outstandingMessages[messageId];
+ } else if (messageId === counter) {
+ handleMessage(message);
+ setSubscribeCounterValue(messageId + 1);
+ subscription.ack(message.ackId);
+ delete outstandingMessages[messageId];
+ } else {
+ // Have not yet processed the message on which this message is dependent
+ return false;
+ }
+ });
+ });
+}
+// [END pubsub_pull_ordered_messages]
+
 // [START pubsub_get_subscription_policy]
 function getSubscriptionPolicy (subscriptionName, callback) {
  // References an existing subscription, e.g. "my-subscription"
@@ -255,6 +311,7 @@ const program = module.exports = {
  deleteSubscription: deleteSubscription,
  getSubscriptionMetadata: getSubscriptionMetadata,
  pullMessages: pullMessages,
+ pullOrderedMessages: pullOrderedMessages,
  getSubscriptionPolicy: getSubscriptionPolicy,
  setSubscriptionPolicy: setSubscriptionPolicy,
  testSubscriptionPermissions: testSubscriptionPermissions,

pubsub/topics.js

@@ -105,6 +105,52 @@ function publishMessage (topicName, data, callback) {
 }
 // [END pubsub_publish_message]
 
+let publishCounterValue = 1;
+
+function getPublishCounterValue () {
+ return publishCounterValue;
+}
+
+function setPublishCounterValue (value) {
+ publishCounterValue = value;
+}
+
+// [START pubsub_publish_ordered_message]
+function publishOrderedMessage (topicName, data, callback) {
+ // References an existing topic, e.g. "my-topic"
+ const topic = pubsubClient.topic(topicName);
+
+ /**
+ * In Node.js, a PubSub message requires a "data" property, which can have a
+ * string or an object as its value. An optional "attributes" property can be
+ * an object of key/value pairs, where the keys and values are both strings.
+ * See https://cloud.google.com/pubsub/reference/rpc/google.pubsub.v1#google.pubsub.v1.PubsubMessage
+ *
+ * Topic#publish() takes either a single message object or an array of message
+ * objects. See https://googlecloudplatform.github.io/google-cloud-node/#/docs/pubsub/latest/pubsub/topic?method=publish
+ */
+ const message = {
+ data: data,
+
+ // Assign an id to the message
+ messageId: getPublishCounterValue()
+ };
+
+ topic.publish(message, (err, messageIds) => {
+ if (err) {
+ callback(err);
+ return;
+ }
+
+ // Update the counter value
+ setPublishCounterValue(message.messageId + 1);
+
+ console.log(`Message ${messageIds[0]} published.`);
+ callback();
+ });
+}
+// [END pubsub_publish_ordered_message]
+
 // [START pubsub_get_topic_policy]
 function getTopicPolicy (topicName, callback) {
  // References an existing topic, e.g. "my-topic"
@@ -191,10 +237,16 @@ const program = module.exports = {
  createTopic: createTopic,
  deleteTopic: deleteTopic,
  publishMessage: publishMessage,
+<<<<<<< 35b153f2b1c74343053f4e9d6dd9951cca326f73
  getTopicPolicy: getTopicPolicy,
  setTopicPolicy: setTopicPolicy,
  testTopicPermissions: testTopicPermissions,
  main: (args) => {
+=======
+ publishOrderedMessage: publishOrderedMessage,
+ listTopics: listTopics,
+ main: function (args) {
+>>>>>>> Initial commit.
  // Run the command-line program
  cli.help().strict().parse(args).argv;
  }