async def _manage_stream(
        self, request_queue: asyncio.Queue[quantum.QuantumRunStreamRequest | None]
    ) -> None:
        """The stream coroutine, an asyncio coroutine to manage QuantumRunStream.

        This coroutine reads responses from the stream and forwards them to the ResponseDemux, where
        the corresponding execution coroutine `_manage_request()` is notified.

        When the stream breaks, the stream is reopened, and all execution coroutines are notified.

        There is at most a single instance of this coroutine running.

        Args:
            request_queue: The queue holding requests from the execution coroutine.
        """
        while True:
            try:
                # The default gRPC client timeout is used.
                response_iterable = await self._grpc_client.quantum_run_stream(
                    _request_iterator(request_queue)
                )
                async for response in response_iterable:
                    self._response_demux.publish(response)
            except asyncio.CancelledError:
                await request_queue.put(None)
                break
            except BaseException as e:
                # Note: the message ID counter is not reset upon a new stream.
                await request_queue.put(None)
                self._response_demux.publish_exception(e)  # Raise to all request tasks
The stream coroutine, an asyncio coroutine to manage QuantumRunStream.

        This coroutine reads responses from the stream and forwards them to the ResponseDemux, where
        the corresponding execution coroutine `_manage_request()` is notified.

        When the stream breaks, the stream is reopened, and all execution coroutines are notified.

        There is at most a single instance of this coroutine running.

        Args:
            request_queue: The queue holding requests from the execution coroutine.
        
cirq-google/cirq_google/engine/stream_manager.py