sp-hydra-veil-core/core/observers/EventBus.py

19 lines
566 B
Python

class EventBus:
_subscribers = {}
@classmethod
def subscribe(cls, event_type, callback):
if event_type not in cls._subscribers:
cls._subscribers[event_type] = []
cls._subscribers[event_type].append(callback)
@classmethod
def emit(cls, event_type, data=None):
for callback in cls._subscribers.get(event_type, []):
callback(data)
@classmethod
def unsubscribe(cls, event_type, callback):
if event_type in cls._subscribers:
cls._subscribers[event_type].remove(callback)