google_cloud_pubsub_publish_and_pull_subscription_quickstart.py

1import os
2from google.cloud import pubsub_v1
3
4# TODO: Set your Google Cloud Project ID and Topic/Subscription IDs
5project_id = "your-project-id"
6topic_id = "your-topic-id"
7subscription_id = "your-subscription-id"
8
9def publish_messages():
10    """Publishes multiple messages to a Pub/Sub topic."""
11    publisher = pubsub_v1.PublisherClient()
12    topic_path = publisher.topic_path(project_id, topic_id)
13
14    for n in range(1, 6):
15        data_str = f"Message number {n}"
16        # Data must be a bytestring
17        data = data_str.encode("utf-8")
18        # When you publish a message, the client returns a future.
19        future = publisher.publish(topic_path, data)
20        print(f"Published message ID: {future.result()}")
21
22def receive_messages():
23    """Receives messages from a pull subscription."""
24    subscriber = pubsub_v1.SubscriberClient()
25    subscription_path = subscriber.subscription_path(project_id, subscription_id)
26
27    def callback(message: pubsub_v1.subscriber.message.Message) -> None:
28        print(f"Received message: {message.data.decode('utf-8')}")
29        message.ack()
30
31    streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
32    print(f"Listening for messages on {subscription_path}..\n")
33
34    # Wrap subscriber in a 'with' block to automatically call close() on termination.
35    with subscriber:
36        try:
37            # When `timeout` is not set, streaming_pull_future.result() will block indefinitely
38            # unless an exception is encountered.
39            streaming_pull_future.result(timeout=10.0)
40        except Exception as e:
41            streaming_pull_future.cancel()  # Trigger the shutdown.
42            streaming_pull_future.result()  # Block until the shutdown is complete.
43
44if __name__ == "__main__":
45    # Note: Ensure the topic and subscription exist before running
46    publish_messages()
47    receive_messages()