async def quantum_run_stream(
        self, requests: AsyncIterator[quantum.QuantumRunStreamRequest], **kwargs
    ) -> Awaitable[AsyncIterable[quantum.QuantumRunStreamResponse]]:
        """Fakes the QuantumRunStream RPC.

        Once a request is received, it is appended to `all_stream_requests`, and the test calling
        `wait_for_requests()` is notified.

        The response is sent when a test calls `reply()` with a `QuantumRunStreamResponse`. If a
        test calls `reply()` with an exception, it is raised here to the `quantum_run_stream()`
        caller.

        This is called from the asyncio thread.
        """
        responses_and_exceptions: asyncio.Queue[
            quantum.QuantumRunStreamResponse | BaseException
        ] = asyncio.Queue()
        self._responses_and_exceptions_future.try_set_result(responses_and_exceptions)

        async def read_requests():
            async for request in requests:
                self.all_stream_requests.append(request)
                self._request_buffer.add(request)
            await responses_and_exceptions.put(FakeQuantumRunStream._REQUEST_STOPPED)
            self._request_iterator_stopped.try_set_result(None)

        async def response_iterator():
            asyncio.create_task(read_requests())
            while (
                message := await responses_and_exceptions.get()
            ) != FakeQuantumRunStream._REQUEST_STOPPED:
                if isinstance(message, quantum.QuantumRunStreamResponse):
                    yield message
                else:  # isinstance(message, BaseException)
                    self._responses_and_exceptions_future = duet.AwaitableFuture()
                    raise message

        return response_iterator()
Fakes the QuantumRunStream RPC.

        Once a request is received, it is appended to `all_stream_requests`, and the test calling
        `wait_for_requests()` is notified.

        The response is sent when a test calls `reply()` with a `QuantumRunStreamResponse`. If a
        test calls `reply()` with an exception, it is raised here to the `quantum_run_stream()`
        caller.

        This is called from the asyncio thread.
        
cirq-google/cirq_google/engine/stream_manager_test.py