google_pubsub_async_subscriber_with_callback_and_timeout.py

1import concurrent.futures
2from google.cloud import pubsub_v1
3
4# TODO(developer)
5# project_id = "your-project-id"
6# subscription_id = "your-subscription-id"
7# Number of seconds the subscriber should listen for messages
8# timeout = 5.0
9
10subscriber = pubsub_v1.SubscriberClient()
11# The `subscription_path` method creates a fully qualified identifier
12# in the form `projects/{project_id}/subscriptions/{subscription_id}`
13subscription_path = subscriber.subscription_path(project_id, subscription_id)
14
15def callback(message: pubsub_v1.subscriber.message.Message) -> None:
16    print(f"Received {message.data!r}.")
17    message.ack()
18
19streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
20print(f"Listening for messages on {subscription_path}..\n")
21
22# Wrap subscriber in a 'with' block to automatically call close() on termination.
23with subscriber:
24    try:
25        # When `timeout` is not None, the polling stops after the specified period.
26        streaming_pull_future.result(timeout=timeout)
27    except concurrent.futures.TimeoutError:
28        streaming_pull_future.cancel()  # Trigger the shutdown.
29        streaming_pull_future.result()  # Block until the shutdown is complete.