fix: closes tailing streams in bidi classes.

Always put `None` into the request queue when closing a bidi stream.

This ensures that the request queue is always signaled as closed, even if the underlying gRPC call object is not yet available.

Author: pujawadare

google/api_core/bidi.py

@@ -281,10 +281,11 @@ def open(self):
 
     def close(self):
         """Closes the stream."""
+        self._request_queue.put(None)
+
         if self.call is None:
             return
 
-        self._request_queue.put(None)
         self.call.cancel()
         self._request_generator = None
         self._initial_request = None

tests/unit/test_bidi.py

@@ -305,6 +305,10 @@ def test_close_no_rpc(self):
         bidi_rpc = bidi.BidiRpc(None)
         bidi_rpc.close()
 
+        # ensure the request queue was signaled to stop.
+        assert bidi_rpc.pending_requests == 1
+        assert bidi_rpc._request_queue.get() is None
+
     def test_send(self):
         rpc, call = make_rpc()
         bidi_rpc = bidi.BidiRpc(rpc)