core/events.py - master - Vibek/First_Game

core/events.py

48 строк2 KB

Vibek

Создана предположительная структура проекта. Реализована первая версия шины событий. Релизован конечный автомат

15 апр 2026, 20:10

15 апр 2026, 20:108244c76

"""
Шина событий (EventBus). Реализует паттерн Observer.
Обеспечивает слабую связанность (Loose Coupling) между модулями.
"""

from typing import Callable, Dict, List, Any, Optional

class EventBus:
    """Singleton шина событий. Гарантирует единственный экземпляр в приложении."""
    
    _instance: Optional['EventBus'] = None
    _listeners: Dict[str, List[Callable[[Any], None]]]

    def __new__(cls) -> 'EventBus':
        """Паттерн Singleton. Создает объект только один раз."""
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._listeners = {}
        return cls._instance

    def subscribe(self, event_name: str, callback: Callable[[Any], None]) -> None:
        """
        Подписать функцию-обработчик на событие.
        :param event_name: Имя события (строка-константа)
        :param callback: Функция, принимающая один аргумент (data)
        """
        if event_name not in self._listeners:
            self._listeners[event_name] = []
        
        if callback not in self._listeners[event_name]:      # Защита от дублирования подписок
            self._listeners[event_name].append(callback)

    def emit(self, event_name: str, data: Any = None) -> None:
        """
        Отправить событие всем подписчикам.
        :param event_name: Имя события
        :param data: Полезная нагрузка (словарь, объект, или None)
        """
        if event_name not in self._listeners:
            return

        for callback in self._listeners[event_name][:]:             # Копия списка [:] защищает от RuntimeError,
            callback(data)                                          # если callback вызовет unsubscribe() во время итерации

    def unsubscribe(self, event_name: str, callback: Callable[[Any], None]) -> None:
        """Отписать функцию от события."""
        if event_name in self._listeners and callback in self._listeners[event_name]:
            self._listeners[event_name].remove(callback)