Example of how to use Google Pub/Sub

google_pub_sub.py

import logging
from google.cloud import pubsub


logger = logging.getLogger(__name__)


def create_topic(topic_name):
    client = pubsub.Client()
    topic = client.topic(topic_name)
    if not topic.exists():
        logger.info("Creating topic {}".format(topic_name))
        topic.create(topic_name)


def create_subscription(topic_name, subscription_name):
    client = pubsub.Client()
    topic = client.topic(topic_name)
    subscription = topic.subscription(subscription_name)
    if not subscription.exists():
        subscription.create()


def publish_message(topic_name, data):
    """Publishes a message to a Pub/Sub topic with the given data."""
    client = pubsub.Client()
    topic = client.topic(topic_name)
    # Data must be a bytestring
    data = data.encode('utf-8')
    message_id = topic.publish(data)
    logger.info('Message {} published.'.format(message_id))


def receive_message(topic_name, subscription_name):
    """Receives a message from a pull subscription.

    `subscription.pull` will return generator of tuples(ack_id, message)
    message.message_id: An ID assigned to the message by the API
    message.data: The payload of the message.
    message.attributes: Extra metadata associated by the publisher with the message.

    :param topic_name: Topic name defined in Google Pub/Sub
    :param subscription_name: Subscription name
    :return: subscription_obj, List(Tuple(ack_id, message))
    """
    client = pubsub.Client()
    topic = client.topic(topic_name)

    # maximum time after a subscriber receives a message before the subscriber should acknowledge the message
    # default is 10 seconds
    # The maximum custom deadline you can specify is 600 seconds (10 minutes)
    subscription = topic.subscription(subscription_name, ack_deadline=600)
    # Change return_immediately=False to block until messages are received.
    results = subscription.pull(return_immediately=False, max_messages=10)
    logging.info('Received {} messages.'.format(len(results)))
    return subscription, results