63 lines
2.1 KiB
Python
63 lines
2.1 KiB
Python
from loguru import logger
|
|
from typing import Dict, List, Type, Callable, Any
|
|
from threading import Lock
|
|
|
|
|
|
class Event:
|
|
"""事件基类"""
|
|
|
|
def __init__(self, **kwargs):
|
|
for key, value in kwargs.items():
|
|
setattr(self, key, value)
|
|
|
|
|
|
class EventManager:
|
|
"""事件管理器，单例模式"""
|
|
_instance = None
|
|
_lock = Lock()
|
|
|
|
@classmethod
|
|
def get_instance(cls):
|
|
"""获取事件管理器实例"""
|
|
if cls._instance is None:
|
|
with cls._lock:
|
|
if cls._instance is None:
|
|
cls._instance = cls()
|
|
return cls._instance
|
|
|
|
def __init__(self):
|
|
if EventManager._instance is not None:
|
|
raise RuntimeError("EventManager 是单例类，请使用 get_instance() 方法获取实例")
|
|
|
|
self.handlers: Dict[Type[Event], List[Callable]] = {}
|
|
self.LOG = logger
|
|
|
|
def register(self, event_type: Type[Event], handler: Callable) -> None:
|
|
"""注册事件处理器"""
|
|
if event_type not in self.handlers:
|
|
self.handlers[event_type] = []
|
|
|
|
if handler not in self.handlers[event_type]:
|
|
self.handlers[event_type].append(handler)
|
|
self.LOG.debug(f"注册事件处理器: {event_type.__name__} -> {handler.__name__}")
|
|
|
|
def unregister(self, event_type: Type[Event], handler: Callable) -> None:
|
|
"""取消注册事件处理器"""
|
|
if event_type in self.handlers and handler in self.handlers[event_type]:
|
|
self.handlers[event_type].remove(handler)
|
|
self.LOG.debug(f"取消注册事件处理器: {event_type.__name__} -> {handler.__name__}")
|
|
|
|
def publish(self, event_type: Type[Event], event_data: Dict[str, Any] = None) -> None:
|
|
"""发布事件"""
|
|
if event_data is None:
|
|
event_data = {}
|
|
|
|
event = event_type(**event_data)
|
|
|
|
if event_type in self.handlers:
|
|
for handler in self.handlers[event_type]:
|
|
try:
|
|
handler(event)
|
|
except Exception as e:
|
|
self.LOG.error(f"事件处理器 {handler.__name__} 处理 {event_type.__name__} 事件出错: {e}")
|