Repository: googleapis/python-pubsub

Issue:
Closed subscriber should raise an error if used as context manager
#### Environment details

  - OS type and version: Linux Ubuntu 20.04
  - Python version: Python 3.8.10
  - pip version: pip 20.0.2 from /usr/lib/python3/dist-packages/pip (python 3.8)
  - `google-cloud-pubsub` version: 
```
Name: google-cloud-pubsub
Version: 2.7.1
Summary: Google Cloud Pub/Sub API client library
Home-page: https://github.com/googleapis/python-pubsub
Author: Google LLC
Author-email: googleapis-packages@google.com
License: Apache 2.0
Location: /usr/local/lib/python3.8/dist-packages
Requires: proto-plus, grpc-google-iam-v1, packaging, google-api-core, libcst, grpcio
Required-by:
```

#### Steps to reproduce

  1. Setup example code as below
  2. Run example code
  
I have also verified that everything is working for me by running the equivalent code in GO and I can read messages from my PubSub subscription OK.

This problem has appeared before here; https://github.com/googleapis/python-pubsub/issues/176  

It seems that it either was never fixed or has come back. 

#### Code example

```python
#------------------------------------------------------------------------
# Example code to create pull subscription to cdr_intl 
# and print the results
#------------------------------------------------------------------------

from google.cloud import pubsub_v1
from concurrent.futures import TimeoutError

project_id = "the-project"
topic_id = "the-topic"
subscription_id = "the-sub"
# Number of seconds the subscriber should listen for messages
timeout = 5.0

publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
topic_path = publisher.topic_path(project_id, topic_id)
# The `subscription_path` method creates a fully qualified identifier
# in the form `projects/{project_id}/subscriptions/{subscription_id}`
subscription_path = subscriber.subscription_path(project_id, subscription_id)

#------------------------------------------------------------------------
# Create the subscription
#------------------------------------------------------------------------

# Wrap the subscriber in a 'with' block to automatically call close() to
# close the underlying gRPC channel when done.
with subscriber:
    try:
        subscription = subscriber.create_subscription(
            request={"name": subscription_path, "topic": topic_path}
        )
        print(f"Subscription created: {subscription}")
    except:
        print(f"Subscription already exists: {subscription_id}")

#------------------------------------------------------------------------
# Read messages
#------------------------------------------------------------------------
def callback(message: pubsub_v1.subscriber.message.Message) -> None:
    print(f"Received {message}.")
    message.ack()

streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
print(f"Listening for messages on {subscription_path}..\n")

# Wrap subscriber in a 'with' block to automatically call close() when done.
with subscriber:
    try:
        # When `timeout` is not set, result() will block indefinitely,
        # unless an exception is encountered first.
        streaming_pull_future.result(timeout=timeout)
    except TimeoutError:
           streaming_pull_future.cancel()  # Trigger the shutdown.
           streaming_pull_future.result()  # Block until the shutdown is complete.
```

#### Stack trace
```
Thread-ConsumeBidirectionalStream caught unexpected exception Cannot invoke RPC: Channel closed! and will exit.
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/dist-packages/google/api_core/bidi.py", line 637, in _thread_main
    self._bidi_rpc.open()
  File "/usr/local/lib/python3.8/dist-packages/google/api_core/bidi.py", line 280, in open
    call = self._start_rpc(iter(request_generator), metadata=self._rpc_metadata)
  File "/usr/local/lib/python3.8/dist-packages/google/pubsub_v1/services/subscriber/client.py", line 1184, in streaming_pull
    response = rpc(requests, retry=retry, timeout=timeout, metadata=metadata,)
  File "/usr/local/lib/python3.8/dist-packages/google/api_core/gapic_v1/method.py", line 145, in __call__
    return wrapped_func(*args, **kwargs)
  File "/usr/local/lib/python3.8/dist-packages/google/api_core/retry.py", line 285, in retry_wrapped_func
    return retry_target(
  File "/usr/local/lib/python3.8/dist-packages/google/api_core/retry.py", line 188, in retry_target
    return target()
  File "/usr/local/lib/python3.8/dist-packages/google/api_core/grpc_helpers.py", line 154, in error_remapped_callable
    result = callable_(*args, **kwargs)
  File "/usr/local/lib/python3.8/dist-packages/grpc/_channel.py", line 1205, in __call__
    call = self._managed_call(
  File "/usr/local/lib/python3.8/dist-packages/grpc/_channel.py", line 1301, in create
    call = state.channel.integrated_call(flags, method, host, deadline,
  File "src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi", line 476, in grpc._cython.cygrpc.Channel.integrated_call
  File "src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi", line 294, in grpc._cython.cygrpc._integrated_call
  File "src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi", line 218, in grpc._cython.cygrpc._call
  File "src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi", line 258, in grpc._cython.cygrpc._call
ValueError: Cannot invoke RPC: Channel closed!
```




Discussion:

Patch:
diff --git a/google/cloud/pubsub_v1/subscriber/client.py b/google/cloud/pubsub_v1/subscriber/client.py
index 85b8800..c4b229a 100644
--- a/google/cloud/pubsub_v1/subscriber/client.py
+++ b/google/cloud/pubsub_v1/subscriber/client.py
@@ -85,6 +85,7 @@ class Client(object):
         # Instantiate the underlying GAPIC client.
         self._api = subscriber_client.SubscriberClient(**kwargs)
         self._target = self._api._transport._host
+        self._closed = False
 
     @classmethod
     def from_service_account_file(cls, filename, **kwargs):
@@ -120,6 +121,14 @@ class Client(object):
         """The underlying gapic API client."""
         return self._api
 
+    @property
+    def closed(self) -> bool:
+        """Return whether the client has been closed and cannot be used anymore.
+
+        .. versionadded:: 2.8.0
+        """
+        return self._closed
+
     def subscribe(
         self,
         subscription,
@@ -252,8 +261,11 @@ class Client(object):
         This method is idempotent.
         """
         self.api._transport.grpc_channel.close()
+        self._closed = True
 
     def __enter__(self):
+        if self._closed:
+            raise RuntimeError("Closed subscriber cannot be used as context manager.")
         return self
 
     def __exit__(self, exc_type, exc_val, exc_tb):


Test patch:
diff --git a/tests/unit/pubsub_v1/subscriber/test_subscriber_client.py b/tests/unit/pubsub_v1/subscriber/test_subscriber_client.py
index 7624c92..601b40b 100644
--- a/tests/unit/pubsub_v1/subscriber/test_subscriber_client.py
+++ b/tests/unit/pubsub_v1/subscriber/test_subscriber_client.py
@@ -50,6 +50,11 @@ def test_init_default_client_info(creds):
         assert expected_client_info in user_agent
 
 
+def test_init_default_closed_state(creds):
+    client = subscriber.Client(credentials=creds)
+    assert not client.closed
+
+
 def test_init_w_custom_transport(creds):
     transport = SubscriberGrpcTransport(credentials=creds)
     client = subscriber.Client(transport=transport)
@@ -185,6 +190,7 @@ def test_close(creds):
         client.close()
 
     patched_close.assert_called()
+    assert client.closed
 
 
 def test_closes_channel_as_context_manager(creds):
@@ -198,6 +204,18 @@ def test_closes_channel_as_context_manager(creds):
     patched_close.assert_called()
 
 
+def test_context_manager_raises_if_closed(creds):
+    client = subscriber.Client(credentials=creds)
+
+    with mock.patch.object(client.api._transport.grpc_channel, "close"):
+        client.close()
+
+    expetect_msg = r"(?i).*closed.*cannot.*context manager.*"
+    with pytest.raises(RuntimeError, match=expetect_msg):
+        with client:
+            pass
+
+
 def test_streaming_pull_gapic_monkeypatch(creds):
     client = subscriber.Client(credentials=creds)