google-cloud-pubsub

import os import time from concurrent.futures import TimeoutError from google.cloud import pubsub_v1 project_id = os.environ.get('GOOGLE_CLOUD_PROJECT') or os.environ.get('GCP_PROJECT') or 'your-gcp-project-id' topic_id = 'my-topic-id' subscription_id = 'my-subscription-id' if not project_id or project_id == 'your-gcp-project-id': raise ValueError("Please set the GOOGLE_CLOUD_PROJECT environment variable or replace 'your-gcp-project-id'.") 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 topic if it doesn't exist try: publisher.get_topic(request={"topic": topic_path}) print(f"Topic {topic_path} already exists.") except Exception: print(f"Creating topic {topic_path}...") publisher.create_topic(request={"name": topic_path}) print(f"Topic {topic_path} created.") # Create subscription if it doesn't exist try: subscriber.get_subscription(request={"subscription": subscription_path}) print(f"Subscription {subscription_path} already exists.") except Exception: print(f"Creating subscription {subscription_path}...") subscriber.create_subscription(request={"name": subscription_path, "topic": topic_path}) print(f"Subscription {subscription_path} created.") # --- Publisher --- message_data = "Hello, Pub/Sub!" print(f"Publishing message: '{message_data}' to {topic_path}") future = publisher.publish(topic_path, message_data.encode('utf-8')) message_id = future.result() print(f"Published message with ID: {message_id}") # --- Subscriber --- def callback(message: pubsub_v1.subscriber.message.Message): print(f"Received message: {message.data.decode('utf-8')}") print(f"Acknowledging message: {message.message_id}") message.ack() print(f"Listening for messages on {subscription_path}...") streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback) # Wrap subscriber in a 'with' block to automatically call close() when done. with subscriber: try: # `subscribe` is non-blocking, so we must keep the main thread from exiting to allow it to run. streaming_pull_future.result(timeout=30) # Wait 30 seconds for messages except TimeoutError: streaming_pull_future.cancel() # Trigger the shutdown. streaming_pull_future.result() # Block until the shutdown is complete. except KeyboardInterrupt: streaming_pull_future.cancel() # Trigger the shutdown. streaming_pull_future.result() # Block until the shutdown is complete. print("Finished listening for messages.") # Clean up resources (optional) # publisher.delete_topic(request={"topic": topic_path}) # subscriber.delete_subscription(request={"subscription": subscription_path}) # print(f"Topic {topic_id} and subscription {subscription_id} deleted.")