Class AEvent

tests/acompat.py:91–102

Subclass of asyncio.Event adding a wait with timeout like threading.Event. wait_timeout() is converted to wait() by async_to_sync.

class AEvent(asyncio.Event):
    """
    Subclass of asyncio.Event adding a wait with timeout like threading.Event.

    wait_timeout() is converted to wait() by async_to_sync.
    """

    async def wait_timeout(self, timeout):
        try:
            return await asyncio.wait_for(self.wait(), timeout)
        except asyncio.TimeoutError:
            return False


class Queue(queue.Queue):  # type: ignore[type-arg]