| import asyncio |
| import logging |
|
|
| logger = logging.getLogger("qkernel.event_bus") |
|
|
|
|
| class EventBus: |
| def __init__(self): |
| self._subscribers: dict[str, list] = {} |
|
|
| def subscribe(self, event_type: str, handler) -> None: |
| if event_type not in self._subscribers: |
| self._subscribers[event_type] = [] |
| self._subscribers[event_type].append(handler) |
|
|
| def unsubscribe(self, event_type: str, handler) -> None: |
| handlers = self._subscribers.get(event_type, []) |
| if handler in handlers: |
| handlers.remove(handler) |
|
|
| async def publish(self, event_type: str, data: dict) -> None: |
| handlers = self._subscribers.get(event_type, []) |
| if not handlers: |
| return |
| logger.debug("Publishing %s to %d handler(s)", event_type, len(handlers)) |
| results = await asyncio.gather( |
| *[handler(data) for handler in handlers], |
| return_exceptions=True, |
| ) |
| for i, r in enumerate(results): |
| if isinstance(r, Exception): |
| logger.error("Handler %d failed on %s: %s", i, event_type, r) |
|
|
