def subscribe(self, message_id: str) -> asyncio.Future:
        """Subscribes to the QuantumRunStreamResponse with a matching ID.

        There should only be one subscriber per message ID.

        Returns:
            A future for the response, to be fulfilled when publish is called.

        Raises:
            ValueError: when trying to subscribe to a message_id which already has a subscriber.
        """

        if message_id in self._subscribers:
            raise ValueError(f'There is already a subscriber for the message ID {message_id}')

        response_future: asyncio.Future = asyncio.get_running_loop().create_future()
        self._subscribers[message_id] = response_future
        return response_future
Subscribes to the QuantumRunStreamResponse with a matching ID.

        There should only be one subscriber per message ID.

        Returns:
            A future for the response, to be fulfilled when publish is called.

        Raises:
            ValueError: when trying to subscribe to a message_id which already has a subscriber.
        
cirq-google/cirq_google/engine/stream_manager.py