Add SubscriberClient.close() to examples

pubsub/cloud-client/iam.py

@@ -51,6 +51,8 @@ def get_subscription_policy(project, subscription_name):
     print("Policy for subscription {}:".format(subscription_path))
     for binding in policy.bindings:
         print("Role: {}, Members: {}".format(binding.role, binding.members))
+
+    client.close()
     # [END pubsub_get_subscription_policy]
 
 
@@ -101,6 +103,8 @@ def set_subscription_policy(project, subscription_name):
             subscription_name, policy
         )
     )
+
+    client.close()
     # [END pubsub_set_subscription_policy]
 
 
@@ -144,6 +148,8 @@ def check_subscription_permissions(project, subscription_name):
             subscription_path, allowed_permissions
         )
     )
+
+    client.close()
     # [END pubsub_test_subscription_permissions]
 
 

pubsub/cloud-client/iam_test.py

@@ -49,7 +49,9 @@ def topic(publisher_client):
 
 @pytest.fixture(scope="module")
 def subscriber_client():
-    yield pubsub_v1.SubscriberClient()
+    subscriber_client = pubsub_v1.SubscriberClient()
+    yield subscriber_client
+    subscriber_client.close()
 
 
 @pytest.fixture

pubsub/cloud-client/quickstart.py

@@ -90,6 +90,9 @@ def callback(message):
         else:
             # Sleeps the thread at 50Hz to save on resources.
             time.sleep(1.0 / 50)
+
+    # Release subscriber's underlying resources.
+    subscriber.close()
     # [END pubsub_end_to_end]
 
 

pubsub/cloud-client/quickstart/sub.py

@@ -48,12 +48,14 @@ def callback(message):
     )
     print("Listening for messages on {}..\n".format(subscription_path))
 
-    # Calling result() on StreamingPullFuture keeps the main thread from
-    # exiting while messages get processed in the callbacks.
-    try:
-        streaming_pull_future.result()
-    except:  # noqa
-        streaming_pull_future.cancel()
+    # Wrap subscriber in a 'with' block to automatically call close() when done.
+    with subscriber:
+        try:
+            # Calling result() on StreamingPullFuture keeps the main thread from
+            # exiting while messages get processed in the callbacks.
+            streaming_pull_future.result()
+        except:  # noqa
+            streaming_pull_future.cancel()
 
 
 if __name__ == "__main__":

pubsub/cloud-client/quickstart/sub_test.py

@@ -62,6 +62,7 @@ def subscription_path(topic_path):
         yield subscription_path
 
     subscriber_client.delete_subscription(subscription_path)
+    subscriber_client.close()
 
 
 def _publish_messages(topic_path):
@@ -102,3 +103,5 @@ def mock_result():
     out, _ = capsys.readouterr()
     assert "Received message" in out
     assert "Acknowledged message" in out
+
+    real_client.close()

pubsub/cloud-client/quickstart_test.py

@@ -66,6 +66,7 @@ def subscription(subscriber_client, topic):
     yield SUBSCRIPTION
 
     subscriber_client.delete_subscription(subscription_path)
+    subscriber_client.close()
 
 
 def test_end_to_end(topic, subscription, capsys):

pubsub/cloud-client/subscriber.py

@@ -52,6 +52,8 @@ def list_subscriptions_in_project(project_id):
 
     for subscription in subscriber.list_subscriptions(project_path):
         print(subscription.name)
+
+    subscriber.close()
     # [END pubsub_list_subscriptions]
 
 
@@ -75,6 +77,8 @@ def create_subscription(project_id, topic_name, subscription_name):
     )
 
     print("Subscription created: {}".format(subscription))
+
+    subscriber.close()
     # [END pubsub_create_pull_subscription]
 
 
@@ -104,6 +108,8 @@ def create_push_subscription(
 
     print("Push subscription created: {}".format(subscription))
     print("Endpoint for subscription is: {}".format(endpoint))
+
+    subscriber.close()
     # [END pubsub_create_push_subscription]
 
 
@@ -123,6 +129,8 @@ def delete_subscription(project_id, subscription_name):
     subscriber.delete_subscription(subscription_path)
 
     print("Subscription deleted: {}".format(subscription_path))
+
+    subscriber.close()
     # [END pubsub_delete_subscription]
 
 
@@ -158,6 +166,8 @@ def update_subscription(project_id, subscription_name, endpoint):
 
     print("Subscription updated: {}".format(subscription_path))
     print("New endpoint for subscription is: {}".format(result.push_config))
+
+    subscriber.close()
     # [END pubsub_update_push_configuration]
 
 
@@ -188,12 +198,14 @@ def callback(message):
     )
     print("Listening for messages on {}..\n".format(subscription_path))
 
-    # result() in a future will block indefinitely if `timeout` is not set,
-    # unless an exception is encountered first.
-    try:
-        streaming_pull_future.result(timeout=timeout)
-    except:  # noqa
-        streaming_pull_future.cancel()
+    # Wrap subscriber in a 'with' block to automatically call close() when done.
+    with subscriber:
+        try:
+            # result() in a future will block indefinitely if `timeout` is not
+            # set, unless an exception is encountered first.
+            streaming_pull_future.result(timeout=timeout)
+        except:  # noqa
+            streaming_pull_future.cancel()
     # [END pubsub_subscriber_async_pull]
     # [END pubsub_quickstart_subscriber]
 
@@ -230,12 +242,14 @@ def callback(message):
     )
     print("Listening for messages on {}..\n".format(subscription_path))
 
-    # result() in a future will block indefinitely if `timeout` is not set,
-    # unless an exception is encountered first.
-    try:
-        streaming_pull_future.result(timeout=timeout)
-    except:  # noqa
-        streaming_pull_future.cancel()
+    # Wrap subscriber in a 'with' block to automatically call close() when done.
+    with subscriber:
+        try:
+            # result() in a future will block indefinitely if `timeout` is not
+            # set, unless an exception is encountered first.
+            streaming_pull_future.result(timeout=timeout)
+        except:  # noqa
+            streaming_pull_future.cancel()
     # [END pubsub_subscriber_async_pull_custom_attributes]
     # [END pubsub_subscriber_sync_pull_custom_attributes]
 
@@ -269,12 +283,14 @@ def callback(message):
     )
     print("Listening for messages on {}..\n".format(subscription_path))
 
-    # result() in a future will block indefinitely if `timeout` is not set,
-    # unless an exception is encountered first.
-    try:
-        streaming_pull_future.result(timeout=timeout)
-    except:  # noqa
-        streaming_pull_future.cancel()
+    # Wrap subscriber in a 'with' block to automatically call close() when done.
+    with subscriber:
+        try:
+            # result() in a future will block indefinitely if `timeout` is not
+            # set, unless an exception is encountered first.
+            streaming_pull_future.result(timeout=timeout)
+        except:  # noqa
+            streaming_pull_future.cancel()
     # [END pubsub_subscriber_flow_settings]
 
 
@@ -309,6 +325,8 @@ def synchronous_pull(project_id, subscription_name):
             len(response.received_messages)
         )
     )
+
+    subscriber.close()
     # [END pubsub_subscriber_sync_pull]
 
 
@@ -398,6 +416,8 @@ def worker(msg):
             len(response.received_messages)
         )
     )
+
+    subscriber.close()
     # [END pubsub_subscriber_sync_pull_with_lease]
 
 
@@ -436,6 +456,8 @@ def callback(message):
                 subscription_name, e
             )
         )
+
+    subscriber.close()
     # [END pubsub_subscriber_error_listener]
 
 

pubsub/cloud-client/subscriber_test.py

@@ -52,7 +52,9 @@ def topic(publisher_client):
 
 @pytest.fixture(scope="module")
 def subscriber_client():
-    yield pubsub_v1.SubscriberClient()
+    subscriber_client = pubsub_v1.SubscriberClient()
+    yield subscriber_client
+    subscriber_client.close()
 
 
 @pytest.fixture(scope="module")