Inform all registered track-event observers about a track change.
This is the subject of the classical observer pattern
Source code in nowplaying/track/handler.py
| class TrackEventHandler:
"""Inform all registered track-event observers about a track change.
This is the subject of the classical observer pattern
"""
def __init__(self: Self) -> None:
"""Initialize the track event handler."""
self.__observers: list[TrackObserver] = []
def register_observer(self: Self, observer: TrackObserver) -> None:
"""Register an observer to be informed about track changes."""
logger.info("Registering TrackObserver '%s'", observer.__class__.__name__)
self.__observers.append(observer)
def remove_observer(self: Self, observer: TrackObserver) -> None:
"""Remove an observer from the list of observers."""
self.__observers.remove(observer)
def get_observers(self: Self) -> list:
"""Return register observers to allow inspecting them."""
return self.__observers
def track_started(self: Self, track: Track) -> None:
"""Inform all registered track-event observers about a track started event."""
logger.info(
"Sending track-started event to %s observers: %s",
len(self.__observers),
track,
)
for observer in self.__observers:
logger.debug(
"Sending track-started event to observer %s",
observer.__class__,
)
try:
observer.track_started(track)
except Exception:
logger.exception(_EXCEPTION_TRACK_HANDLER_ERROR_START)
def track_finished(self: Self, track: Track) -> None:
"""Inform all registered track-event observers about a track finished event."""
logger.info(
"Sending track-finished event to %s observers: %s",
len(self.__observers),
track,
)
for observer in self.__observers:
logger.debug(
"Sending track-finished event to observer %s",
observer.__class__,
)
try:
observer.track_finished(track)
except Exception:
logger.exception(_EXCEPTION_TRACK_HANDLER_ERROR_FINISH)
|
__init__()
Initialize the track event handler.
Source code in nowplaying/track/handler.py
| def __init__(self: Self) -> None:
"""Initialize the track event handler."""
self.__observers: list[TrackObserver] = []
|
get_observers()
Return register observers to allow inspecting them.
Source code in nowplaying/track/handler.py
| def get_observers(self: Self) -> list:
"""Return register observers to allow inspecting them."""
return self.__observers
|
register_observer(observer)
Register an observer to be informed about track changes.
Source code in nowplaying/track/handler.py
| def register_observer(self: Self, observer: TrackObserver) -> None:
"""Register an observer to be informed about track changes."""
logger.info("Registering TrackObserver '%s'", observer.__class__.__name__)
self.__observers.append(observer)
|
remove_observer(observer)
Remove an observer from the list of observers.
Source code in nowplaying/track/handler.py
| def remove_observer(self: Self, observer: TrackObserver) -> None:
"""Remove an observer from the list of observers."""
self.__observers.remove(observer)
|
track_finished(track)
Inform all registered track-event observers about a track finished event.
Source code in nowplaying/track/handler.py
| def track_finished(self: Self, track: Track) -> None:
"""Inform all registered track-event observers about a track finished event."""
logger.info(
"Sending track-finished event to %s observers: %s",
len(self.__observers),
track,
)
for observer in self.__observers:
logger.debug(
"Sending track-finished event to observer %s",
observer.__class__,
)
try:
observer.track_finished(track)
except Exception:
logger.exception(_EXCEPTION_TRACK_HANDLER_ERROR_FINISH)
|
track_started(track)
Inform all registered track-event observers about a track started event.
Source code in nowplaying/track/handler.py
| def track_started(self: Self, track: Track) -> None:
"""Inform all registered track-event observers about a track started event."""
logger.info(
"Sending track-started event to %s observers: %s",
len(self.__observers),
track,
)
for observer in self.__observers:
logger.debug(
"Sending track-started event to observer %s",
observer.__class__,
)
try:
observer.track_started(track)
except Exception:
logger.exception(_EXCEPTION_TRACK_HANDLER_ERROR_START)
|