onyx_online/vban-cmd-python
1
0
Fork 0
mirror of https://github.com/onyx-and-iris/vban-cmd-python.git synced 2025-01-18 18:40:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

subject class extended to support callbacks

Browse Source
This commit is contained in:
onyx-and-iris 2023-06-25 03:41:10 +01:00
parent 8485992495
commit 9196a4e267
1 changed files with 41 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
vban_cmd/subject.py
View File

@ -1,15 +1,14 @@
import logging import logging
logger = logging.getLogger(__name__)
class Subject: class Subject:
"""Adds support for observers"""
logger = logging.getLogger("subject.subject")
def __init__(self): def __init__(self):
"""list of current observers""" """Adds support for observers and callbacks"""
self._observers = list() self._observers = list()
self.logger = logger.getChild(self.__class__.__name__)
@property @property
def observers(self) -> list: def observers(self) -> list:
@ -17,38 +16,57 @@ class Subject:
return self._observers return self._observers
def notify(self, modifier=None): def notify(self, event):
"""run callbacks on update""" """run callbacks on update"""
[o.on_update(modifier) for o in self._observers] for o in self._observers:
if hasattr(o, "on_update"):
o.on_update(event)
else:
if o.__name__ == f"on_{event}":
o()
def add(self, observer): def add(self, observer):
"""adds an observer to _observers""" """adds an observer to observers"""
try:
iterator = iter(observer)
for o in iterator:
if o not in self._observers:
self._observers.append(o)
self.logger.info(f"{o} added to event observers")
else:
self.logger.error(f"Failed to add {o} to event observers")
except TypeError:
if observer not in self._observers: if observer not in self._observers:
self._observers.append(observer) self._observers.append(observer)
self.logger.info(f"{type(observer).__name__} added to event observers") self.logger.info(f"{observer} added to event observers")
else: else:
self.logger.error( self.logger.error(f"Failed to add {observer} to event observers")
f"Failed to add {type(observer).__name__} to event observers"
)
register = add register = add
def remove(self, observer): def remove(self, observer):
"""removes an observer from _observers""" """removes an observer from observers"""
try: try:
self._observers.remove(observer) iterator = iter(observer)
self.logger.info(f"{type(observer).__name__} removed from event observers") for o in iterator:
try:
self._observers.remove(o)
self.logger.info(f"{o} removed from event observers")
except ValueError: except ValueError:
self.logger.error( self.logger.error(f"Failed to remove {o} from event observers")
f"Failed to remove {type(observer).__name__} from event observers" except TypeError:
) try:
self._observers.remove(observer)
self.logger.info(f"{observer} removed from event observers")
except ValueError:
self.logger.error(f"Failed to remove {observer} from event observers")
deregister = remove deregister = remove
def clear(self): def clear(self):
"""clears the _observers list""" """clears the observers list"""
self._observers.clear() self._observers.clear()