Method wait_for_message

tests/test_asyncio/test_cluster.py:4692–4721

Helper method to wait for a message with timeout. Args: pubsub: The PubSub instance timeout: Timeout in seconds ignore_subscribe_messages: Whether to ignore subscribe messages sharded: If True, use get_sharded_message() instead of get_message(

(
        self, pubsub, timeout=0.2, ignore_subscribe_messages=False, sharded=False
    )

    """

    async def wait_for_message(
        self, pubsub, timeout=0.2, ignore_subscribe_messages=False, sharded=False
    ):
        """Helper method to wait for a message with timeout.

        Args:
            pubsub: The PubSub instance
            timeout: Timeout in seconds
            ignore_subscribe_messages: Whether to ignore subscribe messages
            sharded: If True, use get_sharded_message() instead of get_message()
        """
        import asyncio

        now = asyncio.get_running_loop().time()
        end_time = now + timeout
        while now < end_time:
            if sharded:
                message = await pubsub.get_sharded_message(
                    ignore_subscribe_messages=ignore_subscribe_messages,
                    timeout=0.01,
                )
            else:
                message = await pubsub.get_message(
                    ignore_subscribe_messages=ignore_subscribe_messages
                )
            if message is not None:
                return message
            await asyncio.sleep(0.01)
            now = asyncio.get_running_loop().time()
        return None

    def make_message(self, type, channel, data, pattern=None):
        """Helper method to create expected message format"""