Messaging with Pub/Sub

Welcome back! So far, you have learned how to work with Google Cloud services like Compute Engine, Cloud Firestore, and Cloud Datastore. In this lesson, we will explore another important Google Cloud service: Pub/Sub (Publish/Subscribe).

In modern cloud applications, different parts of your system often need to communicate with each other. Sometimes, you want to send a message from one part of your application to another, but you do not want them to be tightly connected. This is where messaging services like Pub/Sub come in. They help you build applications that are more flexible, reliable, and easier to scale.

By the end of this lesson, you will know how to:

• Create a Pub/Sub topic and a subscription.
• Publish messages to a Pub/Sub topic (acting as a producer).
• Receive and process messages from a subscription (acting as a consumer).
• Publish results to another Pub/Sub topic.

Let’s get started!

The first step is to create the messaging resources: a Pub/Sub topic and a subscription. The topic is where messages will be published, and the subscription is how messages are received.

Let’s start by importing the necessary libraries and creating unique names for our resources:

import uuid
from google.cloud import pubsub_v1

project_id = "demo-project"
topic_id = f"orders-{uuid.uuid4().hex[:8]}"
subscription_id = f"orders-sub-{uuid.uuid4().hex[:8]}"

publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()

topic_path = publisher.topic_path(project_id, topic_id)
subscription_path = subscriber.subscription_path(project_id, subscription_id)

# Create the topic
publisher.create_topic(request={"name": topic_path})

# Create the subscription
subscriber.create_subscription(request={"name": subscription_path, "topic": topic_path})

print("Created Topic:", topic_path)
print("Created Subscription:", subscription_path)

• publisher.create_topic(...) creates a new Pub/Sub topic.
• subscriber.create_subscription(...) creates a new subscription to the topic.
• We use uuid to generate unique names for the topic and subscription.

Example Output:

Created Topic: projects/demo-project/topics/orders-1a2b3c4d
Created Subscription: projects/demo-project/subscriptions/orders-sub-5e6f7g8h

Now you have a Pub/Sub topic and a subscription ready to use!

Next, let’s act as a producer and publish messages to the Pub/Sub topic. In this example, we will send two simple order messages.

First, let’s prepare the messages and publish them to the topic:

import json

orders = [
    {"order_id": "o-001", "amount": 42.0},
    {"order_id": "o-002", "amount": 99.5}
]

for o in orders:
    data = json.dumps(o).encode("utf-8")
    future = publisher.publish(topic_path, data)
    print("Published:", o["order_id"])

• We create a list of orders, each represented as a dictionary.
• We use json.dumps(o).encode("utf-8") to convert each order to a JSON string and encode it as bytes, which is the format Pub/Sub expects.
• publisher.publish(topic_path, data) publishes the message to the topic.

Example Output:

Published: o-001
Published: o-002

This is how you publish messages to a Pub/Sub topic. The producer part is now complete.

Now, let’s act as a consumer. We will pull messages from the Pub/Sub subscription, process them, and then acknowledge them so they are not delivered again.

Here’s how you can receive and process messages:

import time

def process_order(body):
    return {"order_id": body["order_id"], "status": "processed", "amount": body["amount"]}

deadline = time.time() + 20  # Run for up to 20 seconds
while time.time() < deadline:
    response = subscriber.pull(
        request={
            "subscription": subscription_path,
            "max_messages": 5,
        },
        timeout=5
    )
    ack_ids = []
    for received_message in response.received_messages:
        body = json.loads(received_message.message.data.decode("utf-8"))
        result = process_order(body)
        print("Processed:", result["order_id"])
        ack_ids.append(received_message.ack_id)
    if ack_ids:
        subscriber.acknowledge(
            request={
                "subscription": subscription_path,
                "ack_ids": ack_ids,
            }
        )
    if not response.received_messages:
        break

• subscriber.pull(...) fetches up to 5 messages from the subscription.
• We use json.loads(...decode("utf-8")) to convert the message back to a dictionary.
• The process_order function simulates processing the order.
• After processing, we call subscriber.acknowledge(...) to acknowledge the messages so they are not delivered again.

Example Output:

Processed: o-001
Processed: o-002

This completes the consumer part. You have now received and processed messages from Pub/Sub.

After processing each order, let’s publish the result to another Pub/Sub topic. This is useful if you want to notify other parts of your system or trigger further actions.

First, create a new topic for processed results:

result_topic_id = f"order-results-{uuid.uuid4().hex[:8]}"
result_topic_path = publisher.topic_path(project_id, result_topic_id)
publisher.create_topic(request={"name": result_topic_path})
print("Created Result Topic:", result_topic_path)

Now, after processing each order, publish the result to the new topic:

for received_message in response.received_messages:
    body = json.loads(received_message.message.data.decode("utf-8"))
    result = process_order(body)
    publisher.publish(result_topic_path, json.dumps(result).encode("utf-8"))
    print("Processed and published:", result["order_id"])
    ack_ids.append(received_message.ack_id)

• publisher.publish(result_topic_path, ...) sends the processed order to the result topic.
• This allows other systems or users subscribed to the topic to receive notifications.

Example Output:

Processed and published: o-001
Processed and published: o-002

Now, every processed order is published to another Pub/Sub topic, making your application more flexible and event-driven.

In this lesson, you learned how to:

• Create a Pub/Sub topic and a subscription.
• Publish messages to a Pub/Sub topic as a producer.
• Receive, process, and acknowledge messages from a subscription as a consumer.
• Publish processed results to another Pub/Sub topic.

This producer/consumer pattern is a powerful way to build decoupled and scalable applications in Google Cloud. You are now ready to practice these skills with hands-on exercises. Try creating your own topics and subscriptions, sending and processing messages, and publishing results to other topics. This will help you become more comfortable with event-driven development in Google Cloud. Good luck!