google_cloud_pubsub_topic_subscription_publish_receive_quickstart.py

1import time
2from google.cloud import pubsub_v1
3
4# TODO: Set these variables before running
5project_id = "your-project-id"
6topic_id = "your-topic-id"
7subscription_id = "your-subscription-id"
8
9# 1. CREATE A TOPIC AND PUBLISH A MESSAGE
10publisher = pubsub_v1.PublisherClient()
11topic_path = publisher.topic_path(project_id, topic_id)
12
13# Create the topic
14try:
15    publisher.create_topic(name=topic_path)
16    print(f"Topic created: {topic_path}")
17except Exception as e:
18    print(f"Topic might already exist: {e}")
19
20# Publish a message
21data = "Hello, World!"
22data = data.encode("utf-8")
23future = publisher.publish(topic_path, data)
24print(f"Published message ID: {future.result()}")
25
26# 2. CREATE A SUBSCRIPTION AND RECEIVE THE MESSAGE
27subscriber = pubsub_v1.SubscriberClient()
28subscription_path = subscriber.subscription_path(project_id, subscription_id)
29
30# Create the subscription
31try:
32    subscriber.create_subscription(name=subscription_path, topic=topic_path)
33    print(f"Subscription created: {subscription_path}")
34except Exception as e:
35    print(f"Subscription might already exist: {e}")
36
37def callback(message: pubsub_v1.subscriber.message.Message) -> None:
38    print(f"Received message: {message.data.decode('utf-8')}")
39    message.ack()
40
41# Subscribe to the topic
42streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
43print(f"Listening for messages on {subscription_path}...\n")
44
45with subscriber:
46    try:
47        # Stop listening after 5 seconds for this quickstart demo
48        streaming_pull_future.result(timeout=5.0)
49    except Exception:
50        streaming_pull_future.cancel()  # Trigger the shutdown.
51        streaming_pull_future.result()  # Block until the shutdown is complete.