feat: closed subscriber as context manager raises (#488)

Closes #484.

**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

@@ -85,6 +85,7 @@ def __init__(self, **kwargs):
         # 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 @@ def api(self):
         """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 @@ def close(self):
         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):

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)