425 lines
13 KiB
Python
425 lines
13 KiB
Python
import asyncio
|
|
import datetime
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
from app.features.core.utils import gen_random
|
|
from app.library.log import get_logger
|
|
|
|
from .BackgroundWorker import BackgroundWorker
|
|
from .Singleton import Singleton
|
|
|
|
LOG = get_logger()
|
|
|
|
|
|
class Events:
|
|
"""
|
|
The events that can be emitted.
|
|
"""
|
|
|
|
STARTUP: str = "startup"
|
|
LOADED: str = "loaded"
|
|
STARTED: str = "started"
|
|
SHUTDOWN: str = "shutdown"
|
|
|
|
CONNECTED: str = "connected"
|
|
|
|
CONFIG_UPDATE: str = "config_update"
|
|
|
|
LOG_INFO: str = "log_info"
|
|
LOG_WARNING: str = "log_warning"
|
|
LOG_ERROR: str = "log_error"
|
|
LOG_SUCCESS: str = "log_success"
|
|
|
|
ITEM_ADDED: str = "item_added"
|
|
ITEM_UPDATED: str = "item_updated"
|
|
ITEM_PROGRESS: str = "item_progress"
|
|
ITEM_COMPLETED: str = "item_completed"
|
|
ITEM_CANCELLED: str = "item_cancelled"
|
|
ITEM_DELETED: str = "item_deleted"
|
|
ITEM_BULK_DELETED: str = "item_bulk_deleted"
|
|
ITEM_PAUSED: str = "item_paused"
|
|
ITEM_RESUMED: str = "item_resumed"
|
|
ITEM_MOVED: str = "item_moved"
|
|
ITEM_STATUS: str = "item_status"
|
|
ITEM_ERROR: str = "item_error"
|
|
|
|
TEST: str = "test"
|
|
ADD_URL: str = "add_url"
|
|
|
|
PAUSED: str = "paused"
|
|
RESUMED: str = "resumed"
|
|
|
|
TASKS_ADD: str = "task_add"
|
|
TASK_DISPATCHED: str = "task_dispatched"
|
|
TASK_FINISHED: str = "task_finished"
|
|
TASK_ERROR: str = "task_error"
|
|
|
|
SCHEDULE_ADD: str = "schedule_add"
|
|
|
|
def get_all() -> list:
|
|
"""
|
|
Get all the events.
|
|
|
|
Returns:
|
|
list: The list of events.
|
|
|
|
"""
|
|
return [
|
|
getattr(Events, ev) for ev in dir(Events) if not ev.startswith("_") and not callable(getattr(Events, ev))
|
|
]
|
|
|
|
def frontend() -> list:
|
|
"""
|
|
Get the frontend events.
|
|
|
|
Returns:
|
|
list: The list of frontend events.
|
|
|
|
"""
|
|
return [
|
|
Events.CONFIG_UPDATE,
|
|
Events.CONNECTED,
|
|
Events.LOG_INFO,
|
|
Events.LOG_WARNING,
|
|
Events.LOG_ERROR,
|
|
Events.LOG_SUCCESS,
|
|
Events.ITEM_ADDED,
|
|
Events.ITEM_UPDATED,
|
|
Events.ITEM_PROGRESS,
|
|
Events.ITEM_CANCELLED,
|
|
Events.ITEM_DELETED,
|
|
Events.ITEM_BULK_DELETED,
|
|
Events.ITEM_MOVED,
|
|
Events.ITEM_STATUS,
|
|
Events.PAUSED,
|
|
Events.RESUMED,
|
|
]
|
|
|
|
def only_debug() -> list:
|
|
"""
|
|
High frequency events that should only be logged in debug mode.
|
|
|
|
Returns:
|
|
list: The list of debug events.
|
|
|
|
"""
|
|
return [Events.ITEM_UPDATED, Events.ITEM_PROGRESS]
|
|
|
|
|
|
@dataclass(kw_only=True)
|
|
class Event:
|
|
"""
|
|
Event is a data transfer object that represents an event that was emitted.
|
|
"""
|
|
|
|
id: str = field(default_factory=lambda: str(gen_random(16)), init=False)
|
|
"""The id of the event."""
|
|
|
|
created_at: str = field(default_factory=lambda: str(datetime.datetime.now(tz=datetime.UTC).isoformat()))
|
|
"""The time the event was created."""
|
|
|
|
event: str
|
|
"""The event that was emitted."""
|
|
|
|
title: str | None = None
|
|
"""The title of the event, if any."""
|
|
|
|
message: str | None = None
|
|
"""The message of the event, if any."""
|
|
|
|
data: Any
|
|
"""The data that was passed to the event."""
|
|
|
|
extras: dict = field(default_factory=dict)
|
|
"""Listeners can add extra data to the event."""
|
|
|
|
def serialize(self) -> dict:
|
|
"""
|
|
Serialize the event.
|
|
|
|
Returns:
|
|
dict: The serialized event.
|
|
|
|
"""
|
|
return {
|
|
"id": self.id,
|
|
"created_at": self.created_at,
|
|
"event": self.event,
|
|
"title": self.title,
|
|
"message": self.message,
|
|
"data": self.data,
|
|
}
|
|
|
|
def __repr__(self) -> str:
|
|
return f"Event(id={self.id}, created_at={self.created_at}, event={self.event}, title={self.title}, message={self.message} data={self.data})"
|
|
|
|
def put(self, key: str, value: Any) -> None:
|
|
"""
|
|
Put extra data to the event.
|
|
|
|
Args:
|
|
key (str): The key of the extra data.
|
|
value (Any): The value of the extra data.
|
|
|
|
"""
|
|
self.extras[key] = value
|
|
|
|
def datatype(self) -> str:
|
|
"""
|
|
Get the datatype of the data.
|
|
|
|
Returns:
|
|
str: The datatype of the data.
|
|
|
|
"""
|
|
return type(self.data).__name__
|
|
|
|
def __str__(self):
|
|
return f"Event(id={self.id}, created_at={self.created_at}, event={self.event}, title={self.title}, message={self.message})"
|
|
|
|
|
|
class EventListener:
|
|
def __init__(self, name: str, callback: Callable[..., Any]):
|
|
self.name: str = name
|
|
"The name of the listener."
|
|
self.call_back: Callable[..., Any] = callback
|
|
"The callback function to call when the event is emitted."
|
|
self.is_coroutine: bool = asyncio.iscoroutinefunction(callback)
|
|
"Whether the callback is a coroutine function or not."
|
|
|
|
async def handle(self, event: Event, **kwargs):
|
|
if self.is_coroutine:
|
|
return self.call_back(event, self.name, **kwargs)
|
|
|
|
return asyncio.create_task(self.call_back(event, self.name, **kwargs), name=f"EL-{self.name}-{event.id}")
|
|
|
|
|
|
class EventBus(metaclass=Singleton):
|
|
"""
|
|
This class is used to subscribe to and emit events to the registered listeners.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self._listeners: dict[str, list[tuple[str, EventListener]]] = {}
|
|
"The listeners for the events."
|
|
|
|
self.debug: bool = False
|
|
"Whether to log debug messages or not."
|
|
|
|
self._offload: BackgroundWorker | None = None
|
|
"The background worker to offload tasks to."
|
|
|
|
@staticmethod
|
|
def get_instance() -> "EventBus":
|
|
"""
|
|
Get the instance of the EventsSubscriber.
|
|
|
|
Returns:
|
|
EventsSubscriber: The instance of the EventsSubscriber
|
|
|
|
"""
|
|
return EventBus()
|
|
|
|
def subscribe(self, event: str | list | tuple, callback: Callable[..., Any], name: str | None = None) -> "EventBus":
|
|
"""
|
|
Subscribe to an event.
|
|
|
|
Args:
|
|
event (str): The event to subscribe to.
|
|
name (str|None): The name of the subscriber, if None a random uuid will be generated.
|
|
callback(Event, name, **kwargs) (Awaitable): The function to call. Must be a coroutine.
|
|
|
|
Returns:
|
|
EventsSubscriber: The instance of the EventsSubscriber
|
|
|
|
"""
|
|
all_events = Events.get_all()
|
|
|
|
if isinstance(event, str):
|
|
if "*" == event:
|
|
event = all_events
|
|
elif "frontend" == event:
|
|
event = Events.frontend()
|
|
else:
|
|
if event not in all_events:
|
|
LOG.error(
|
|
"Listener '%s' tried to subscribe to unknown event '%s'.",
|
|
name,
|
|
event,
|
|
extra={"listener": name, "event": event},
|
|
)
|
|
return self
|
|
|
|
event = [event]
|
|
|
|
if not name:
|
|
name = gen_random(12)
|
|
|
|
for e in event:
|
|
if e not in all_events:
|
|
LOG.error(
|
|
"Listener '%s' tried to subscribe to unknown event '%s'.",
|
|
name,
|
|
e,
|
|
extra={"listener": name, "event": e},
|
|
)
|
|
continue
|
|
|
|
if e not in self._listeners:
|
|
self._listeners[e] = []
|
|
|
|
self._listeners[e] = [(n, listener) for n, listener in self._listeners[e] if n != name]
|
|
self._listeners[e].append((name, EventListener(name, callback)))
|
|
|
|
LOG.debug("Listener '%s' has subscribed.", name, extra={"listener": name, "events": event})
|
|
|
|
return self
|
|
|
|
def unsubscribe(self, event: str | list | tuple, name: str) -> "EventBus":
|
|
"""
|
|
Unsubscribe from an event.
|
|
|
|
Args:
|
|
event (str): The event to unsubscribe from.
|
|
name (str): The name of the subscriber.
|
|
|
|
Returns:
|
|
EventsSubscriber: The instance of the EventsSubscriber
|
|
|
|
"""
|
|
if isinstance(event, str):
|
|
event = [event]
|
|
|
|
events = []
|
|
for e in event:
|
|
if e in self._listeners:
|
|
original_len: int = len(self._listeners[e])
|
|
self._listeners[e] = [(n, listener) for n, listener in self._listeners[e] if n != name]
|
|
if len(self._listeners[e]) < original_len:
|
|
events.append(e)
|
|
|
|
if len(events) > 0:
|
|
LOG.debug("Listener '%s' unsubscribed from '%s'.", name, events, extra={"listener": name, "events": events})
|
|
|
|
return self
|
|
|
|
def emit(
|
|
self, event: str, data: Any | None = None, title: str | None = None, message: str | None = None, **kwargs
|
|
) -> None:
|
|
"""
|
|
Emit an event to all registered listeners.
|
|
|
|
Args:
|
|
event (str): The event to emit.
|
|
data (Any|None): The data to pass to the event.
|
|
title (str | None): The title of the event, if any.
|
|
message (str | None): The message of the event, if any.
|
|
**kwargs: The keyword arguments to pass to the event.
|
|
|
|
"""
|
|
if event not in self._listeners:
|
|
return
|
|
|
|
if not data:
|
|
data = {}
|
|
|
|
ev = Event(event=event, title=title, message=message, data=data)
|
|
|
|
if self.debug or event not in Events.only_debug():
|
|
LOG.debug(
|
|
"Delivering '%s' event to %s listener(s).",
|
|
ev.event,
|
|
len(self._listeners[event]),
|
|
extra={
|
|
"event_id": ev.id,
|
|
"event": ev.event,
|
|
"handler_count": len(self._listeners[event]),
|
|
"data": data,
|
|
},
|
|
)
|
|
|
|
try:
|
|
loop = asyncio.get_running_loop()
|
|
|
|
async def execute_handlers():
|
|
for _, handler in self._listeners[event]:
|
|
try:
|
|
if handler.is_coroutine:
|
|
coro = handler.call_back(ev, handler.name, **kwargs)
|
|
if asyncio.iscoroutine(coro):
|
|
await coro
|
|
else:
|
|
LOG.warning(
|
|
"Async handler '%s' returned '%s' instead of a coroutine.",
|
|
handler.name,
|
|
type(coro).__name__,
|
|
extra={"handler": handler.name, "returned_type": type(coro).__name__},
|
|
)
|
|
else:
|
|
await self._call(handler, ev, kwargs)
|
|
except Exception as e:
|
|
LOG.exception(
|
|
"Failed to emit '%s' event to listener '%s'.",
|
|
ev.event,
|
|
handler.name,
|
|
extra={
|
|
"event_id": ev.id,
|
|
"event": ev.event,
|
|
"handler": handler.name,
|
|
"exception_type": type(e).__name__,
|
|
},
|
|
)
|
|
|
|
loop.create_task(execute_handlers())
|
|
except RuntimeError:
|
|
LOG.debug(
|
|
"No event loop detected; offloading '%s' event to %s background listener(s).",
|
|
ev.event,
|
|
len(self._listeners[event]),
|
|
extra={"event_id": ev.id, "event": ev.event, "handler_count": len(self._listeners[event])},
|
|
)
|
|
for _, handler in self._listeners[event]:
|
|
try:
|
|
if not self._offload:
|
|
self._offload = BackgroundWorker.get_instance()
|
|
|
|
self._offload.submit(handler.handle, ev, **kwargs)
|
|
except Exception as e:
|
|
LOG.exception(
|
|
"Failed to offload '%s' event to listener '%s'.",
|
|
ev.event,
|
|
handler.name,
|
|
extra={
|
|
"event_id": ev.id,
|
|
"event": ev.event,
|
|
"handler": handler.name,
|
|
"exception_type": type(e).__name__,
|
|
},
|
|
)
|
|
|
|
def clear(self) -> None:
|
|
"""
|
|
Clear all listeners. Useful for testing.
|
|
"""
|
|
self._listeners.clear()
|
|
|
|
def debug_enable(self) -> None:
|
|
"""
|
|
Enable debug logging.
|
|
"""
|
|
self.debug = True
|
|
|
|
def debug_disable(self) -> None:
|
|
"""
|
|
Disable debug logging.
|
|
"""
|
|
self.debug = False
|
|
|
|
@staticmethod
|
|
def _call(h, event, kw):
|
|
async def call_handler():
|
|
return h.call_back(event, h.name, **kw)
|
|
|
|
return call_handler()
|