dispatcher.wait_for future handling race condition about hikari HOT 2 CLOSED

import asyncio
import logging
import selectors

import hikari.impl.event_manager_base


logging.basicConfig(level="DEBUG")


class FooEvent(hikari.Event):

    @property
    def app(self) -> hikari.RESTAware:
        return NotImplemented


class Dispatcher(hikari.impl.event_manager_base.EventManagerBase):
    async def on_foo(self, _, __):
        await self.dispatch(FooEvent())


instance = Dispatcher(NotImplemented, None)


async def producer():
    i = 0
    while True:
        # Do a really short sleep, which puts an event on the selector and 
        # causes a different task to be invoked. Asyncio otherwise appears
        # to immediately invoke the same task, meaning we would keep producing
        # and never get a chance to consume. This might be a CPython bug, I
        # am not sure.

        # Note, setting to anything NON ZERO (even 0.0000000001) seems to hide
        # this bug, annoyingly.
        await asyncio.sleep(0)

        i += 1
        if not (i % 1000):
            print("produced", i)

        instance.consume_raw_event(NotImplemented, "FOO", {})


async def consumer():
    i = 0
    while True:
        i += 1
        if not (i % 1000):
            print("consumed", i)

        def pred(curry):
            def predicate(event):
                nonlocal curry
                if curry:
                    return True
                else:
                    curry = True
                    return False


        await asyncio.gather(
            instance.wait_for(FooEvent, timeout=None, predicate=pred(True)),
            instance.wait_for(FooEvent, timeout=None, predicate=pred(False)),
            instance.wait_for(FooEvent, timeout=None, predicate=pred(True)),
            instance.wait_for(FooEvent, timeout=None, predicate=pred(False)),
        )



async def runner():
    asyncio.create_task(consumer())
    asyncio.create_task(producer())


selector = selectors.SelectSelector()
asyncio.set_event_loop(asyncio.SelectorEventLoop(selector))
asyncio.get_event_loop().set_debug(True)
asyncio.get_event_loop().run_until_complete(runner())
asyncio.get_event_loop().run_forever()