Spaces:
Sleeping
Sleeping
File size: 1,404 Bytes
4f6ee5b 6c9722e 4f6ee5b 6c9722e 4f6ee5b 6c9722e 4f6ee5b 6c9722e 4f6ee5b 6c9722e 4f6ee5b 6c9722e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 |
import logging
setup_logging = __import__('logging_config').setup_logging
setup_logging()
logger = logging.getLogger('channels')
class Event:
def __init__(self):
self.listeners = {}
def on(self, event: str, listener: callable):
if event not in self.listeners:
self.listeners[event] = []
self.listeners[event].append(listener)
logger.info(f"Registered listener for event: {event}")
def off(self, event: str, listener: callable):
if event in self.listeners:
self.listeners[event] = [l for l in self.listeners[event] if l != listener]
logger.info(f"Unregistered listener for event: {event}")
def emit(self, event: str, *args, **kwargs):
if event in self.listeners:
for listener in self.listeners[event]:
try:
listener(*args, **kwargs)
logger.info(f"Emitted event: {event}")
except Exception as e:
logger.error(f"Error emitting event {event}: {e}")
class Channel:
def __init__(self):
self.event = Event()
def on(self, event: str, listener: callable):
self.event.on(event, listener)
def off(self, event: str, listener: callable):
self.event.off(event, listener)
def emit(self, event: str, *args, **kwargs):
self.event.emit(event, *args, **kwargs) |