docs: block until the streaming pull shuts down

google/cloud/pubsub_v1/subscriber/client.py

@@ -188,7 +188,8 @@ def callback(message):
             try:
                 future.result()
             except KeyboardInterrupt:
-                future.cancel()
+                future.cancel()  # Trigger the shutdown.
+                future.result()  # Block until the shutdown is complete.
 
         Args:
             subscription (str): The name of the subscription. The

samples/snippets/quickstart/sub.py

@@ -43,7 +43,8 @@ def callback(message):
         # exiting while messages get processed in the callbacks.
         streaming_pull_future.result(timeout=timeout)
     except:  # noqa
-        streaming_pull_future.cancel()
+        streaming_pull_future.cancel()  # Trigger the shutdown.
+        streaming_pull_future.result()  # Block until the shutdown is complete.
 
     subscriber_client.close()
 

samples/snippets/schema.py

@@ -343,7 +343,8 @@ def callback(message):
             # unless an exception occurs first.
             streaming_pull_future.result(timeout=timeout)
         except TimeoutError:
-            streaming_pull_future.cancel()
+            streaming_pull_future.cancel()  # Trigger the shutdown.
+            streaming_pull_future.result()  # Block until the shutdown is complete.
     # [END pubsub_subscribe_avro_records]
 
 
@@ -393,7 +394,8 @@ def callback(message):
             # unless an exception occurs first.
             streaming_pull_future.result(timeout=timeout)
         except TimeoutError:
-            streaming_pull_future.cancel()
+            streaming_pull_future.cancel()  # Trigger the shutdown.
+            streaming_pull_future.result()  # Block until the shutdown is complete.
     # [END pubsub_subscribe_proto_messages]
 
 

samples/snippets/subscriber.py

@@ -397,7 +397,8 @@ def callback(message):
             # unless an exception is encountered first.
             streaming_pull_future.result(timeout=timeout)
         except TimeoutError:
-            streaming_pull_future.cancel()
+            streaming_pull_future.cancel()  # Trigger the shutdown.
+            streaming_pull_future.result()  # Block until the shutdown is complete.
     # [END pubsub_subscriber_async_pull]
     # [END pubsub_quickstart_subscriber]
 
@@ -436,7 +437,8 @@ def callback(message):
             # unless an exception is encountered first.
             streaming_pull_future.result(timeout=timeout)
         except TimeoutError:
-            streaming_pull_future.cancel()
+            streaming_pull_future.cancel()  # Trigger the shutdown.
+            streaming_pull_future.result()  # Block until the shutdown is complete.
     # [END pubsub_subscriber_async_pull_custom_attributes]
 
 
@@ -474,7 +476,8 @@ def callback(message):
             # unless an exception is encountered first.
             streaming_pull_future.result(timeout=timeout)
         except TimeoutError:
-            streaming_pull_future.cancel()
+            streaming_pull_future.cancel()  # Trigger the shutdown.
+            streaming_pull_future.result()  # Block until the shutdown is complete.
     # [END pubsub_subscriber_flow_settings]
 
 
@@ -663,10 +666,11 @@ def callback(message):
         try:
             streaming_pull_future.result(timeout=timeout)
         except Exception as e:
-            streaming_pull_future.cancel()
             print(
                 f"Listening for messages on {subscription_path} threw an exception: {e}."
             )
+            streaming_pull_future.cancel()  # Trigger the shutdown.
+            streaming_pull_future.result()  # Block until the shutdown is complete.
     # [END pubsub_subscriber_error_listener]
 
 
@@ -697,7 +701,8 @@ def callback(message):
         try:
             streaming_pull_future.result(timeout=timeout)
         except TimeoutError:
-            streaming_pull_future.cancel()
+            streaming_pull_future.cancel()  # Trigger the shutdown.
+            streaming_pull_future.result()  # Block until the shutdown is complete.
     # [END pubsub_dead_letter_delivery_attempt]