GCP PubSub Helper Library

pubsub.js

const { PubSub } = require('@google-cloud/pubsub');

function psInit() {
  const projectId = process.env['GCP_PROJECT'];
  // Instantiates a client
  const pubsub = new PubSub({ projectId });
  return pubsub;
}
async function psPublish(pubsub, topic, data) {
  try {
    const messageId = await pubsub.topic(topic).publishMessage({ json: data });
    console.log(`Message ${messageId} published.`);
  } catch (error) {
    console.error(`Received error while publishing: ${error.message}`);
    process.exitCode = 1;
  }
}

function psSubscribe(pubsub, subscriptionTopic, ifAck, messageFunc) {
  const subscriptionName = subscriptionTopic;
  const maxInProgress = 1; // configure this as user needs

  // Instantiates a client
  const subscriberOptions = {
    flowControl: {
      maxMessages: maxInProgress,
    },
  };

  // References an existing subscription.
  // Note that flow control settings are not persistent across subscribers.
  const subscription = pubsub.subscription(
    subscriptionName,
    subscriberOptions
  );

  console.log(
    `Subscriber to subscription ${subscription.name} is ready to receive messages at a controlled volume of ${maxInProgress} messages.`
  );

  subscription.on('message', messageFunc);
  return `subscribed to ${subscriptionTo}`;
}

module.exports = {
  psInit,
  psPublish,
  psSubscribe,
};