event_emitter.py

from asyncio import Queue
from typing import Callable, TypeVar, Generic, AsyncIterable


T = TypeVar("T")


class EventEmitter(Generic[T]):
    listeners: set[Callable[[T], None]] = set()

    def add_listener(self, listener: Callable[[T], None]):
        self.listeners.add(listener)

    def remove_listener(self, listener: Callable[[T], None]):
        self.listeners.remove(listener)

    def emit(self, data: T):
        for listener in self.listeners:
            try:
                listener(data)
            except:
                pass


async def to_async_iterable(emitter: EventEmitter[T]) -> AsyncIterable[T]:
    queue: Queue[T] = Queue()

    def listener(data: T):
        queue.put_nowait(data)

    emitter.add_listener(listener)

    try:
        while True:
            yield await queue.get()
    finally:
        emitter.remove_listener(listener)