docs: block until the streaming pull shuts down (#424)

Fixes #423.

If subscriber client is used as a context manager, we need to block until the shutdown is complete before leaving the `with` block. See the issue description for more details.

**PR checklist:**
- [x] Make sure to open an issue as a [bug/issue](https://github.com/googleapis/python-pubsub/issues/new/choose) before writing your code!  That way we can discuss the change, evaluate designs, and agree on the general idea
- [x] Ensure the tests and linter pass
- [x] Code coverage does not decrease (if any source code was changed)
- [x] Appropriate docs were updated (if necessary)

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]