25 lines
776 B
Python
25 lines
776 B
Python
from collections import defaultdict
|
|
from typing import Any, Callable
|
|
|
|
from domain.ports import EventBusPort
|
|
|
|
|
|
class EventBus(EventBusPort):
|
|
"""Prosta implementacja szyny zdarzeń (event bus).
|
|
|
|
Zgodnie z lekcją — architektura oparta o zdarzenia umożliwia:
|
|
- monitorowanie działań agenta
|
|
- podejmowanie akcji (np. kompresja kontekstu, logowanie)
|
|
- subskrypcję zdarzeń z różnych komponentów
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._listeners: dict[str, list[Callable]] = defaultdict(list)
|
|
|
|
def emit(self, event: str, data: dict[str, Any]) -> None:
|
|
for callback in self._listeners[event]:
|
|
callback(data)
|
|
|
|
def on(self, event: str, callback: Callable) -> None:
|
|
self._listeners[event].append(callback)
|