google-cloud-pubsublite

from google.cloud.pubsublite.cloudpubsub import PublisherClient

from google.cloud.pubsublite.cloudpubsub import SubscriberClient

from google.cloud.pubsublite.admin import AdminClient

from google.cloud.pubsublite.types import TopicPath

from google.cloud.pubsublite.types import FlowControlSettings

import os import time from concurrent.futures import TimeoutError from google.cloud.pubsublite.cloudpubsub import PublisherClient, SubscriberClient from google.cloud.pubsublite.types import ( CloudZone, TopicPath, SubscriptionPath, FlowControlSettings, ) # Configuration: Ensure GOOGLE_CLOUD_PROJECT and GOOGLE_CLOUD_ZONE are set # or replace placeholders with your actual GCP Project ID and a Lite zone (e.g., 'us-central1-a'). project_id = os.environ.get("GOOGLE_CLOUD_PROJECT", "your-gcp-project-id") zone = os.environ.get("GOOGLE_CLOUD_ZONE", "us-central1-a") topic_id = "my-pubsublite-topic" subscription_id = "my-pubsublite-subscription" # Define topic and subscription paths topic_path = TopicPath(project_id, CloudZone(zone), topic_id) subscription_path = SubscriptionPath(project_id, CloudZone(zone), subscription_id) if project_id == "your-gcp-project-id": print("Please set the GOOGLE_CLOUD_PROJECT and GOOGLE_CLOUD_ZONE environment variables.") print("Alternatively, replace 'your-gcp-project-id' and 'us-central1-a' with your actual values.") exit(1) print(f"Using Topic: {topic_path}") print(f"Using Subscription: {subscription_path}") print("Ensure the topic and subscription already exist in your Pub/Sub Lite project.") # --- Publisher: Send a message --- print("\n--- Publishing messages ---") publisher = PublisherClient() message_data = b"Hello from Pub/Sub Lite Python client!" try: # publish returns a future; .result() waits for acknowledgement future = publisher.publish(topic_path, message_data) message_id = future.result(timeout=5) print(f"Published message '{message_data.decode()}' with ID: {message_id}") except TimeoutError: print("Publishing timed out.") except Exception as e: print(f"Error publishing message: {e}") finally: publisher.close() # Important to close the client to flush messages print("Publisher client closed.") # Allow a moment for the message to propagate time.sleep(2) # --- Subscriber: Receive messages --- print("\n--- Subscribing to messages ---") subscriber = SubscriberClient() # Configure flow control to limit outstanding messages and bytes flow_control_settings = FlowControlSettings( messages_outstanding=10, bytes_outstanding=10 * 1024 * 1024 # 10 MiB ) received_messages = [] def callback(message): payload = message.data.decode() print(f"Received message: '{payload}' (ID: {message.message_id})") received_messages.append(payload) message.ack() # Acknowledge the message to allow it to be discarded # The subscribe method returns a streaming_future that can be cancelled. streaming_future = subscriber.subscribe(subscription_path, callback, flow_control_settings) print(f"Listening for messages on {subscription_path} for 10 seconds...") try: # Wait for the future to complete (e.g., if cancelled) or timeout. streaming_future.result(timeout=10) except TimeoutError: print("No more messages received within 10 seconds or subscription completed.") except Exception as e: print(f"Error during subscription: {e}") finally: streaming_future.cancel() # Stop the subscription stream subscriber.close() # Important to close the client to release resources print("Subscriber client closed.") if received_messages: print(f"\nSuccessfully processed {len(received_messages)} message(s).") else: print("\nNo messages were processed by the subscriber.")