fix: closes tailing streams in bidi classes. (#851)

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,11 +281,11 @@ def open(self):
 
 defclose(self):
 """Closes the stream."""
-ifself.callisNone:
-return
+ifself.callisnotNone:
+self.call.cancel()
 
+# Put None in request queue to signal termination.
 self._request_queue.put(None)
-self.call.cancel()
 self._request_generator=None
 self._initial_request=None
 self._callbacks= []

google/api_core/bidi_async.py

@@ -197,11 +197,11 @@ async def open(self) -> None:
 
 asyncdefclose(self) ->None:
 """Closes the stream."""
-ifself.callisNone:
-return
+ifself.callisnotNone:
+self.call.cancel()
 
+# Put None in request queue to signal termination.
 awaitself._request_queue.put(None)
-self.call.cancel()
 self._request_generator=None
 self._initial_request=None
 self._callbacks= []

tests/asyncio/test_bidi_async.py

@@ -255,6 +255,21 @@ async def test_close(self):
 assertbidi_rpc._initial_requestisNone
 assertnotbidi_rpc._callbacks
 
+@pytest.mark.asyncio
+asyncdeftest_close_with_no_rpc(self):
+bidi_rpc=bidi_async.AsyncBidiRpc(None)
+
+awaitbidi_rpc.close()
+
+assertbidi_rpc.callisNone
+assertbidi_rpc.is_activeisFalse
+# ensure the request queue was signaled to stop.
+assertbidi_rpc.pending_requests==1
+assertawaitbidi_rpc._request_queue.get() isNone
+# ensure request and callbacks are cleaned up
+assertbidi_rpc._initial_requestisNone
+assertnotbidi_rpc._callbacks
+
 @pytest.mark.asyncio
 asyncdeftest_close_no_rpc(self):
 bidi_rpc=bidi_async.AsyncBidiRpc(None)

tests/unit/test_bidi.py

@@ -301,10 +301,19 @@ def test_close(self):
 assertbidi_rpc._initial_requestisNone
 assertnotbidi_rpc._callbacks
 
-deftest_close_no_rpc(self):
+deftest_close_with_no_rpc(self):
 bidi_rpc=bidi.BidiRpc(None)
 bidi_rpc.close()
 
+assertbidi_rpc.callisNone
+assertbidi_rpc.is_activeisFalse
+# ensure the request queue was signaled to stop.
+assertbidi_rpc.pending_requests==1
+assertbidi_rpc._request_queue.get() isNone
+# ensure request and callbacks are cleaned up
+assertbidi_rpc._initial_requestisNone
+assertnotbidi_rpc._callbacks
+
 deftest_send(self):
 rpc, call=make_rpc()
 bidi_rpc=bidi.BidiRpc(rpc)