From a973ce8bcf0c045dcfa34b397c2beebd72e10973 Mon Sep 17 00:00:00 2001 From: ArabCoders Date: Fri, 21 Mar 2025 21:54:28 +0300 Subject: [PATCH] Finalize EventBus design --- app/library/DownloadQueue.py | 2 +- app/library/Emitter.py | 2 +- .../{EventsSubscriber.py => Events.py} | 122 ++++++++++++------ app/library/HttpAPI.py | 2 +- app/library/HttpSocket.py | 12 +- app/library/Notifications.py | 2 +- app/library/Presets.py | 9 +- app/library/Scheduler.py | 14 +- app/library/Tasks.py | 7 +- app/main.py | 6 +- 10 files changed, 105 insertions(+), 73 deletions(-) rename app/library/{EventsSubscriber.py => Events.py} (55%) diff --git a/app/library/DownloadQueue.py b/app/library/DownloadQueue.py index c6f491ce..611bb520 100644 --- a/app/library/DownloadQueue.py +++ b/app/library/DownloadQueue.py @@ -18,7 +18,7 @@ from .config import Config from .DataStore import DataStore from .Download import Download from .Emitter import Emitter -from .EventsSubscriber import Events +from .Events import Events from .ItemDTO import ItemDTO from .Presets import Presets from .Singleton import Singleton diff --git a/app/library/Emitter.py b/app/library/Emitter.py index 8348ab70..6405606a 100644 --- a/app/library/Emitter.py +++ b/app/library/Emitter.py @@ -2,7 +2,7 @@ import asyncio import logging from collections.abc import Awaitable -from .EventsSubscriber import Events +from .Events import Events from .Singleton import Singleton LOG = logging.getLogger("Emitter") diff --git a/app/library/EventsSubscriber.py b/app/library/Events.py similarity index 55% rename from app/library/EventsSubscriber.py rename to app/library/Events.py index 755fd471..c00d6354 100644 --- a/app/library/EventsSubscriber.py +++ b/app/library/Events.py @@ -1,7 +1,9 @@ import asyncio +import datetime import logging +import uuid from collections.abc import Awaitable -from dataclasses import dataclass +from dataclasses import dataclass, field from .Singleton import Singleton @@ -51,11 +53,50 @@ class Events: @dataclass(kw_only=True) class Event: - id: str - data: dict + """ + Event is a data transfer object that represents an event that was emitted. + """ + + id: str = field(default_factory=lambda: str(uuid.uuid4()), init=False) + """The id of the event.""" + + created: str = field(default_factory=lambda: str(datetime.datetime.now(tz=datetime.timezone.UTC).isoformat())) + """The time the event was created.""" + + event: str + """The event that was emitted.""" + + data: any + """The data that was passed to the event.""" + + def serialize(self) -> dict: + """ + Serialize the event. + + Returns: + dict: The serialized event. + + """ + return {"id": self.id, "created": self.created, "event": self.event, "data": self.data} + + def __repr__(self): + return f"Event(id={self.id}, created={self.created}, event={self.event}, data={self.data})" + + 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={self.created}, event={self.event})" -class EventsSubscriber(metaclass=Singleton): +class EventBus(metaclass=Singleton): """ This class is used to subscribe to and emit events to the registered listeners. """ @@ -67,10 +108,10 @@ class EventsSubscriber(metaclass=Singleton): """The listeners for the events.""" def __init__(self): - EventsSubscriber._instance = self + EventBus._instance = self @staticmethod - def get_instance() -> "EventsSubscriber": + def get_instance() -> "EventBus": """ Get the instance of the EventsSubscriber. @@ -78,18 +119,19 @@ class EventsSubscriber(metaclass=Singleton): EventsSubscriber: The instance of the EventsSubscriber """ - if not EventsSubscriber._instance: - EventsSubscriber._instance = EventsSubscriber() - return EventsSubscriber._instance + if not EventBus._instance: + EventBus._instance = EventBus() - def subscribe(self, event: str | list | tuple, id: str, callback: Awaitable) -> "EventsSubscriber": + return EventBus._instance + + def subscribe(self, event: str | list | tuple, callback: Awaitable, name: str | None = None) -> "EventBus": """ Subscribe to an event. Args: event (str): The event to subscribe to. - id (str|None): The id of the subscriber, if None a random uuid will be generated. - callback (Awaitable): The function to call. Must be a coroutine. + name (str|None): The name of the subscriber, if None a random uuid will be generated. + callback(Event) (Awaitable): The function to call. Must be a coroutine. Returns: EventsSubscriber: The instance of the EventsSubscriber @@ -98,21 +140,24 @@ class EventsSubscriber(metaclass=Singleton): if isinstance(event, str): event = [event] + if not name: + name = str(uuid.uuid4()) + for e in event: if e not in self._listeners: self._listeners[e] = {} - self._listeners[e][id] = callback + self._listeners[e][name] = callback return self - def unsubscribe(self, event: str | list | tuple, id: str) -> "EventsSubscriber": + def unsubscribe(self, event: str | list | tuple, name: str) -> "EventBus": """ Unsubscribe from an event. Args: event (str): The event to unsubscribe from. - id (str): The id of the subscriber. + name (str): The name of the subscriber. Returns: EventsSubscriber: The instance of the EventsSubscriber @@ -122,18 +167,18 @@ class EventsSubscriber(metaclass=Singleton): event = [event] for e in event: - if e in self._listeners and id in self._listeners[e]: - del self._listeners[e][id] + if e in self._listeners and name in self._listeners[e]: + del self._listeners[e][name] return self - def emit_sync(self, event: str, *args, **kwargs): + def emit_sync(self, event: str, data: any, **kwargs) -> list: """ Emit an event synchronously. Args: event (str): The event to emit. - *args: The arguments to pass to the event. + data (any): The data to pass to the event. **kwargs: The keyword arguments to pass to the event. Returns: @@ -144,28 +189,26 @@ class EventsSubscriber(metaclass=Singleton): if event not in self._listeners: return [] - results = [] - for id, callback in self._listeners[event].items(): - try: - if "data" not in kwargs or not isinstance(kwargs["data"], Event): - data = Event(id=id, data={"args": args if args else [], **kwargs}) - else: - data = kwargs["data"] + ev = Event(event=event, data=data) + LOG.debug(f"Emitting event '{ev}'.") - results.append(asyncio.get_event_loop().run_until_complete(callback(event, data))) + results = [] + for name, callback in self._listeners[event].items(): + try: + results.append(asyncio.get_event_loop().run_until_complete(callback(ev, name, **kwargs))) except Exception as e: LOG.exception(e) - LOG.error(f"Failed to emit event '{event}' to '{id}'. Error message '{e!s}'.") + LOG.error(f"Failed to emit event '{event}' to '{name}'. Error message '{e!s}'.") return results - def emit(self, event: str, *args, **kwargs): + def emit(self, event: str, data: any, **kwargs) -> Awaitable: """ Emit an event. Args: event (str): The event to emit. - *args: The arguments to pass to the event. + data (any): The data to pass to the event. **kwargs: The keyword arguments to pass to the event. Returns: @@ -175,19 +218,16 @@ class EventsSubscriber(metaclass=Singleton): if event not in self._listeners: return None - tasks = [] - for id, callback in self._listeners[event].items(): - try: - if args and isinstance(args[0], Event): - data = args[0] - elif "data" in kwargs and isinstance(kwargs["data"], Event): - data = kwargs["data"] - else: - data = Event(id=id, data={"args": args if args else [], **kwargs}) + ev = Event(event=event, data=data) - tasks.append(asyncio.create_task(callback(event, data))) + LOG.debug(f"Emitting event '{ev}'.") + + tasks = [] + for name, callback in self._listeners[event].items(): + try: + tasks.append(asyncio.create_task(callback(ev, name, **kwargs))) except Exception as e: LOG.exception(e) - LOG.error(f"Failed to emit event '{event}' to '{id}'. Error message '{e!s}'.") + LOG.error(f"Failed to emit event '{event}' to '{name}'. Error message '{e!s}'.") return asyncio.gather(*tasks) diff --git a/app/library/HttpAPI.py b/app/library/HttpAPI.py index d8454a85..a39c9f47 100644 --- a/app/library/HttpAPI.py +++ b/app/library/HttpAPI.py @@ -26,7 +26,7 @@ from .config import Config from .DownloadQueue import DownloadQueue from .Emitter import Emitter from .encoder import Encoder -from .EventsSubscriber import Events +from .Events import Events from .ffprobe import ffprobe from .M3u8 import M3u8 from .Notifications import Notification, NotificationEvents diff --git a/app/library/HttpSocket.py b/app/library/HttpSocket.py index 30ca13fc..d55c32c1 100644 --- a/app/library/HttpSocket.py +++ b/app/library/HttpSocket.py @@ -18,7 +18,7 @@ from .config import Config from .DownloadQueue import DownloadQueue from .Emitter import Emitter from .encoder import Encoder -from .EventsSubscriber import Event, Events, EventsSubscriber +from .Events import EventBus, Events from .Presets import Presets from .Utils import arg_converter, is_downloaded @@ -80,13 +80,9 @@ class HttpSocket(Common): if hasattr(method, "_ws_event") and self.sio: self.sio.on(method._ws_event)(method) # type: ignore - # self.sio.on("*", es.emit) - - async def handle_event(_: str, data: Event): - LOG.debug(f"Event received. '{data}'") - await self.add(**data.data) - - EventsSubscriber.get_instance().subscribe(Events.ADD_URL, "socket_add_url", handle_event) + EventBus.get_instance().subscribe( + Events.ADD_URL, lambda data, _: self.add(**data.data), "socket_add_url" + ) # register the shutdown event. app.on_shutdown.append(self.on_shutdown) diff --git a/app/library/Notifications.py b/app/library/Notifications.py index f5d0304e..794d7175 100644 --- a/app/library/Notifications.py +++ b/app/library/Notifications.py @@ -10,7 +10,7 @@ import httpx from .config import Config from .encoder import Encoder -from .EventsSubscriber import Events +from .Events import Events from .ItemDTO import ItemDTO from .Singleton import Singleton from .Utils import ag, validate_uuid diff --git a/app/library/Presets.py b/app/library/Presets.py index d9aa29e2..2e0d54d4 100644 --- a/app/library/Presets.py +++ b/app/library/Presets.py @@ -10,7 +10,7 @@ from aiohttp import web from .config import Config from .Emitter import Emitter from .encoder import Encoder -from .EventsSubscriber import Event, Events, EventsSubscriber +from .Events import EventBus, Events from .Singleton import Singleton LOG = logging.getLogger("presets") @@ -81,13 +81,12 @@ class Presets(metaclass=Singleton): except Exception: pass - def handle_event(_, e: Event): - self.save(**e.data) - with open(os.path.join(os.path.dirname(__file__), "presets.json")) as f: self._default_presets = [Preset(**preset) for preset in json.load(f)] - EventsSubscriber.get_instance().subscribe(Events.PRESETS_ADD, f"{__class__}.save", handle_event) + EventBus.get_instance().subscribe( + Events.PRESETS_ADD, lambda data, _: self.add(**data.data), f"{__class__}.save" + ) @staticmethod def get_instance() -> "Presets": diff --git a/app/library/Scheduler.py b/app/library/Scheduler.py index 215bb1d8..7b6756de 100644 --- a/app/library/Scheduler.py +++ b/app/library/Scheduler.py @@ -4,7 +4,7 @@ import logging from aiocron import Cron from aiohttp import web -from .EventsSubscriber import Event, Events, EventsSubscriber +from .Events import EventBus, Events from .Singleton import Singleton LOG = logging.getLogger("scheduler") @@ -26,10 +26,9 @@ class Scheduler(metaclass=Singleton): self._loop = loop or asyncio.get_event_loop() - def handle_event(_, e: Event): - self.add(**e.data) - - EventsSubscriber.get_instance().subscribe(Events.SCHEDULE_ADD, f"{__class__}.add", handle_event) + EventBus.get_instance().subscribe( + Events.SCHEDULE_ADD, lambda data, _: self.add(**data.data), f"{__class__}.add" + ) @staticmethod def get_instance() -> "Scheduler": @@ -76,8 +75,9 @@ class Scheduler(metaclass=Singleton): """Return the job by id.""" return self._jobs.get(id) - def add(self, timer: str, func: callable, args: tuple = (), - kwargs: dict | None = None, id: str | None = None) -> str: + def add( + self, timer: str, func: callable, args: tuple = (), kwargs: dict | None = None, id: str | None = None + ) -> str: """ Add a job to the schedule. diff --git a/app/library/Tasks.py b/app/library/Tasks.py index 0550590b..9d69112d 100644 --- a/app/library/Tasks.py +++ b/app/library/Tasks.py @@ -13,7 +13,7 @@ from aiohttp import web from .config import Config from .Emitter import Emitter from .encoder import Encoder -from .EventsSubscriber import Event, Events, EventsSubscriber +from .Events import Event, EventBus, Events from .Scheduler import Scheduler from .Singleton import Singleton @@ -82,10 +82,7 @@ class Tasks(metaclass=Singleton): except Exception: pass - def handle_event(_, e: Event): - self.save(**e.data) - - EventsSubscriber.get_instance().subscribe(Events.TASKS_ADD, f"{__class__}.save", handle_event) + EventBus.get_instance().subscribe(Events.TASKS_ADD, lambda data, _: self.add(**data.data), f"{__class__}.save") @staticmethod def get_instance() -> "Tasks": diff --git a/app/main.py b/app/main.py index 8262428f..bbe54962 100644 --- a/app/main.py +++ b/app/main.py @@ -12,7 +12,7 @@ from aiohttp import web from library.config import Config from library.DownloadQueue import DownloadQueue from library.Emitter import Emitter -from library.EventsSubscriber import Events, EventsSubscriber +from library.Events import EventBus, Events from library.HttpAPI import HttpAPI from library.HttpSocket import HttpSocket from library.Notifications import Notification @@ -61,7 +61,7 @@ class Main: self._socket = HttpSocket(queue=self._queue) Emitter.get_instance().add_emitter([Notification().emit], local=False).add_emitter( - [EventsSubscriber().emit], local=True + [EventBus().emit], local=True ) self._app.on_cleanup.append(_close_connection) @@ -94,7 +94,7 @@ class Main: """ Start the application. """ - EventsSubscriber.get_instance().emit(Events.STARTUP, data={"app": self._app}) + EventBus.get_instance().emit(Events.STARTUP, data={"app": self._app}) self._socket.attach(self._app) self._http.attach(self._app)