updated the app to use EventBus

app/library/Download.py

@@ -12,7 +12,7 @@ import yt_dlp
 
 from .AsyncPool import Terminator
 from .config import Config
-from .Emitter import Emitter
+from .Events import EventBus, Events
 from .ffprobe import ffprobe
 from .ItemDTO import ItemDTO
 from .YTDLPOpts import YTDLPOpts
@@ -52,7 +52,6 @@ class Download:
     default_ytdl_opts: dict = None
     debug: bool = False
     temp_path: str = None
-    emitter: Emitter = None
     cancelled: bool = False
     is_live: bool = False
     info_dict: dict = None
@@ -102,7 +101,7 @@ class Download:
         self.tmpfilename = None
         self.status_queue = None
         self.proc = None
-        self.emitter = None
+        self._notify = EventBus.get_instance()
         self.max_workers = int(config.max_workers)
         self.temp_keep = bool(config.temp_keep)
         self.is_live = bool(info.is_live) or info.live_in is not None
@@ -222,8 +221,7 @@ class Download:
 
         self.logger.info(f'Task id="{self.info.id}" PID="{os.getpid()}" title="{self.info.title}" completed.')
 
-    async def start(self, emitter: Emitter):
+    async def start(self):
-        self.emitter = emitter
         self.status_queue = Config.get_manager().Queue()
 
         # Create temp dir for each download.
@@ -236,7 +234,7 @@ class Download:
         self.proc.start()
         self.info.status = "preparing"
 
-        asyncio.create_task(self.emitter.updated(dl=self.info), name=f"emitter-{self.id}")
+        await self._notify.emit(Events.UPDATED, data=self.info)
         asyncio.create_task(self.progress_update(), name=f"update-{self.id}")
 
         return await asyncio.get_running_loop().run_in_executor(None, self.proc.join)
@@ -365,7 +363,7 @@ class Download:
                 self.logger.debug(f"Status Update: {self.info._id=} {status=}")
 
             if isinstance(status, str):
-                asyncio.create_task(self.emitter.updated(dl=self.info), name=f"emitter-u-{self.id}")
+                await self._notify.emit(Events.UPDATED, data=self.info)
                 continue
 
             self.tmpfilename = status.get("tmpfilename")
@@ -384,9 +382,7 @@ class Download:
 
             if "error" == self.info.status and "error" in status:
                 self.info.error = status.get("error")
-                asyncio.create_task(
+                await self._notify.emit(Events.ERROR, data={"message": self.info.error, "data": self.info})
-                    self.emitter.error(message=self.info.error, data=self.info), name=f"emitter-e-{self.id}"
-                )
 
             if "downloaded_bytes" in status:
                 total = status.get("total_bytes") or status.get("total_bytes_estimate")
@@ -422,4 +418,4 @@ class Download:
                     self.logger.exception(e)
                     self.logger.error(f"Failed to ffprobe: {status.get}. {e}")
 
-            asyncio.create_task(self.emitter.updated(dl=self.info), name=f"emitter-u-{self.id}")
+            await self._notify.emit(Events.UPDATED, data=self.info)

app/library/DownloadQueue.py

@@ -17,8 +17,7 @@ from .AsyncPool import AsyncPool
 from .config import Config
 from .DataStore import DataStore
 from .Download import Download
-from .Emitter import Emitter
+from .Events import EventBus, Events
-from .Events import Events
 from .ItemDTO import ItemDTO
 from .Presets import Presets
 from .Singleton import Singleton
@@ -53,11 +52,11 @@ class DownloadQueue(metaclass=Singleton):
     _instance = None
     """Instance of the DownloadQueue."""
 
-    def __init__(self, connection: Connection, emitter: Emitter | None = None, config: Config | None = None):
+    def __init__(self, connection: Connection, config: Config | None = None):
         DownloadQueue._instance = self
 
         self.config = config or Config.get_instance()
-        self.emitter = emitter or Emitter.get_instance()
+        self._notify = EventBus.get_instance()
         self.done = DataStore(type=DownloadQueue.TYPE_DONE, connection=connection)
         self.queue = DataStore(type=DownloadQueue.TYPE_QUEUE, connection=connection)
         self.done.load()
@@ -326,9 +325,7 @@ class DownloadQueue(metaclass=Singleton):
                 itemDownload = self.queue.put(dlInfo)
                 self.event.set()
 
-            asyncio.create_task(
+            await self._notify.emit(NotifyEvent, data=itemDownload.info.serialize())
-                self.emitter.emit(NotifyEvent, itemDownload.info), name=f"notifier-{NotifyEvent}-{itemDownload.info.id}"
-            )
 
             return {"status": "ok"}
 
@@ -505,11 +502,11 @@ class DownloadQueue(metaclass=Singleton):
                 await item.close()
                 LOG.debug(f"Deleting from queue {item_ref}")
                 self.queue.delete(id)
-                asyncio.create_task(self.emitter.cancelled(dl=item.info.serialize()), name=f"notifier-c-{id}")
+                await self._notify.emit(Events.CANCELLED, info=item.info.serialize())
                 item.info.status = "cancelled"
                 item.info.error = "Cancelled by user."
                 self.done.put(item)
-                asyncio.create_task(self.emitter.completed(dl=item.info.serialize()), name=f"notifier-d-{id}")
+                await self._notify.emit(Events.COMPLETED, data=item.info.serialize())
                 LOG.info(f"Deleted from queue {item_ref}")
 
             status[id] = "ok"
@@ -563,7 +560,8 @@ class DownloadQueue(metaclass=Singleton):
                     LOG.error(f"Unable to remove '{itemRef}' local file '{filename}'. {e!s}")
 
             self.done.delete(id)
-            asyncio.create_task(self.emitter.cleared(dl=item.info.serialize()), name=f"notifier-c-{id}")
+            await self._notify.emit(Events.CLEARED, data=item.info.serialize())
+
             msg = f"Deleted completed download '{itemRef}'."
             if fileDeleted and filename:
                 msg += f" and removed local file '{filename}'."
@@ -672,7 +670,7 @@ class DownloadQueue(metaclass=Singleton):
 
         try:
             self._active_downloads[entry.info._id] = entry
-            await entry.start(self.emitter)
+            await entry.start()
 
             if "finished" != entry.info.status:
                 if entry.tmpfilename and os.path.isfile(entry.tmpfilename):
@@ -694,12 +692,12 @@ class DownloadQueue(metaclass=Singleton):
             self.queue.delete(key=id)
 
             if entry.is_cancelled() is True:
-                asyncio.create_task(self.emitter.cancelled(dl=entry.info.serialize()), name=f"notifier-c-{id}")
+                await self._notify.emit(Events.CANCELLED, data=entry.info.serialize())
                 entry.info.status = "cancelled"
                 entry.info.error = "Cancelled by user."
 
             self.done.put(value=entry)
-            asyncio.create_task(self.emitter.completed(dl=entry.info.serialize()), name=f"notifier-d-{id}")
+            await self._notify.emit(Events.COMPLETED, data=entry.info.serialize())
         else:
             LOG.warning(f"Download '{id}' not found in queue.")
 

app/library/Emitter.py

@@ -1,132 +0,0 @@
-import asyncio
-import logging
-from collections.abc import Awaitable
-
-from .Events import Events
-from .Singleton import Singleton
-
-LOG = logging.getLogger("Emitter")
-
-
-class Emitter(metaclass=Singleton):
-    """
-    This class is used to emit events to the registered emitters.
-    """
-
-    emitters: list[(str, Awaitable)] = []
-    """The emitters for the events."""
-
-    _instance = None
-
-    def __init__(self):
-        Emitter._instance = self
-
-    @staticmethod
-    def get_instance():
-        """
-        Get the instance of the Emitter.
-
-        Returns:
-            Emitter: The instance of the Emitter
-
-        """
-        if not Emitter._instance:
-            Emitter._instance = Emitter()
-        return Emitter._instance
-
-    def add_emitter(self, emitter: list[Awaitable] | Awaitable, local: bool = False) -> "Emitter":
-        """
-        Add an emitter to the list of emitters.
-
-        Args:
-            emitter (Awaitable|list[Awaitable]): The emitter function. The function must return a coroutine or None.
-            local (bool): Mark the emitter as target for local events.
-
-        Returns:
-            Emitter: The instance of the Emitter
-
-        """
-        if not isinstance(emitter, list):
-            emitter = [emitter]
-
-        for e in emitter:
-            self.emitters.append((local, e))
-
-        return self
-
-    async def added(self, dl: dict, local: bool = False, **kwargs):
-        await self.emit(Events.ADDED, data=dl, local=local, **kwargs)
-
-    async def updated(self, dl: dict, local: bool = False, **kwargs):
-        await self.emit(Events.UPDATED, data=dl, local=local, **kwargs)
-
-    async def completed(self, dl: dict, local: bool = False, **kwargs):
-        await self.emit(Events.COMPLETED, data=dl, local=local, **kwargs)
-
-    async def cancelled(self, dl: dict, local: bool = False, **kwargs):
-        await self.emit(Events.CANCELLED, data=dl, local=local, **kwargs)
-
-    async def cleared(self, dl: dict | None = None, local: bool = False, **kwargs):
-        await self.emit(Events.CLEARED, data=dl, local=local, **kwargs)
-
-    async def error(self, message: str, data: dict|None = None, local: bool = False, **kwargs):
-        msg = {"type": "error", "message": message, "data": {}}
-        if data:
-            msg.update({"data": data})
-        await self.emit(Events.ERROR, data=msg, local=local, **kwargs)
-
-    async def info(self, message: str, data: dict|None = None, local: bool = False, **kwargs):
-        msg = {"type": "info", "message": message, "data": {}}
-        if data:
-            msg.update({"data": data})
-        await self.emit(Events.LOG_INFO, data=msg, local=local, **kwargs)
-
-    async def success(self, message: str, data: dict|None = None, local: bool = False, **kwargs):
-        msg = {"type": "success", "message": message, "data": {}}
-        if data:
-            msg.update({"data": data})
-        await self.emit(Events.LOG_SUCCESS, data=msg, local=local, **kwargs)
-
-    async def emit(self, event: str, data, local: bool = False, **kwargs):
-        """
-        Emit an event.
-
-        Args:
-            event (str): The event to emit.
-            data (dict): The data to send with the event.
-            local (bool): If the event should be sent to the local emitters only.
-            kwargs: Additional arguments to pass to the emitters.
-
-        Returns:
-            None
-
-        """
-        tasks = []
-
-        for emitter in self.emitters:
-            try:
-                isLocal, callback = emitter
-                if local and not isLocal:
-                    continue
-
-                _ret = callback(event, data, **kwargs)
-                if _ret:
-                    if isinstance(_ret, list):
-                        tasks.extend(_ret)
-                    else:
-                        tasks.append(_ret)
-            except Exception as e:
-                LOG.error(f"Emitter '{callback}' failed with error '{e}'.")
-
-        if len(tasks) < 1:
-            return
-
-        try:
-            await asyncio.wait_for(asyncio.gather(*tasks), timeout=60)
-        except asyncio.CancelledError:
-            LOG.error(f"Cancelled sending event '{event}'.")
-        except TimeoutError:
-            LOG.error(f"Timed out sending event '{event}'.")
-        except Exception as e:
-            LOG.exception(e)
-            LOG.error(f"Failed to send event '{event}'. '{e!s}'.")

app/library/Events.py

@@ -10,12 +10,74 @@ from .Singleton import Singleton
 LOG = logging.getLogger("EventsSubscriber")
 
 
+def error(msg: str, data: dict | None = None) -> dict:
+    """
+    Create an error message.
+
+    Args:
+        msg (str): The message.
+        data (dict|None): The data to include in the message.
+
+    Returns:
+        dict : The message wrapped in a dictionary.
+
+    """
+    return message("error", msg, data)
+
+
+def info(msg: str, data: dict | None = None) -> dict:
+    """
+    Create an info message.
+
+    Args:
+        msg (str): The message.
+        data (dict|None): The data to include in the message.
+
+    Returns:
+        dict : The message wrapped in a dictionary.
+
+    """
+    return message("info", msg, data)
+
+
+def success(msg: str, data: dict | None = None) -> dict:
+    """
+    Create a success message.
+
+    Args:
+        msg (str): The message.
+        data (dict|None): The data to include in the message.
+
+    Returns:
+        dict : The message wrapped in a dictionary.
+
+    """
+    return message("success", msg, data)
+
+
+def message(type: str, message: str, data: dict | None = None) -> dict:
+    """
+    Create a message.
+
+    Args:
+        type (str): The type of the message.
+        message (str): The message.
+        data (dict|None): The data to include in the message.
+
+    Returns:
+        dict : The message wrapped in a dictionary.
+
+    """
+    return {"type": type, "message": message, "data": data if data else {}}
+
+
 class Events:
     """
     The events that can be emitted.
     """
 
     STARTUP = "startup"
+    LOADED = "loaded"
     SHUTDOWN = "shutdown"
 
     ADDED = "added"
@@ -50,6 +112,46 @@ class Events:
     PRESETS_UPDATE = "presets_update"
     SCHEDULE_ADD = "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.ADDED,
+            Events.UPDATED,
+            Events.COMPLETED,
+            Events.CANCELLED,
+            Events.CLEARED,
+            Events.ERROR,
+            Events.LOG_INFO,
+            Events.LOG_SUCCESS,
+            Events.INITIAL_DATA,
+            Events.YTDLP_CONVERT,
+            Events.ITEM_DELETE,
+            Events.ITEM_CANCEL,
+            Events.STATUS,
+            Events.CLI_CLOSE,
+            Events.CLI_OUTPUT,
+            Events.UPDATE,
+            Events.PAUSED,
+        ]
+
 
 @dataclass(kw_only=True)
 class Event:
@@ -60,7 +162,7 @@ class Event:
     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()))
+    created: str = field(default_factory=lambda: str(datetime.datetime.now(tz=datetime.timezone.utc).isoformat()))
     """The time the event was created."""
 
     event: str
@@ -96,6 +198,23 @@ class Event:
         return f"Event(id={self.id}, created={self.created}, event={self.event})"
 
 
+class EventListener:
+    name: str
+    is_coroutine: bool = False
+    call_back: callable
+
+    def __init__(self, name: str, callback: callable):
+        self.name = name
+        self.call_back = callback
+        self.is_coroutine = asyncio.iscoroutinefunction(callback)
+
+    async def handle(self, event: Event, **kwargs):
+        if self.is_coroutine:
+            return self.call_back(event, self.name, **kwargs)
+        else:
+            return asyncio.create_task(self.call_back(event, self.name, **kwargs))
+
+
 class EventBus(metaclass=Singleton):
     """
     This class is used to subscribe to and emit events to the registered listeners.
@@ -104,7 +223,7 @@ class EventBus(metaclass=Singleton):
     _instance = None
     """the instance of the EventsSubscriber"""
 
-    _listeners: dict[str, list[str, Awaitable]] = {}
+    _listeners: dict[str, list[str, EventListener]] = {}
     """The listeners for the events."""
 
     def __init__(self):
@@ -137,17 +256,32 @@ class EventBus(metaclass=Singleton):
             EventsSubscriber: The instance of the EventsSubscriber
 
         """
+        all_events = Events.get_all()
+
         if isinstance(event, str):
-            event = [event]
+            if "*" == event:
+                event = all_events
+            elif "frontend" == event:
+                event = Events.frontend()
+            else:
+                if event not in all_events:
+                    LOG.error(f"'{name}' attempted to listen on '{event}' which does not exist.")
+                    return self
+
+                event = [event]
 
         if not name:
             name = str(uuid.uuid4())
 
         for e in event:
+            if e not in all_events:
+                LOG.error(f"'{name}' attempted to listen on '{e}' which does not exist.")
+                continue
+
             if e not in self._listeners:
                 self._listeners[e] = {}
 
-            self._listeners[e][name] = callback
+            self._listeners[e][name] = EventListener(name, callback)
 
         return self
 
@@ -172,7 +306,7 @@ class EventBus(metaclass=Singleton):
 
         return self
 
-    def emit_sync(self, event: str, data: any, **kwargs) -> list:
+    def sync_emit(self, event: str, data: any, **kwargs) -> list:
         """
         Emit an event synchronously.
 
@@ -190,19 +324,19 @@ class EventBus(metaclass=Singleton):
             return []
 
         ev = Event(event=event, data=data)
-        LOG.debug(f"Emitting event '{ev}'.")
+        LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data})
 
         results = []
-        for name, callback in self._listeners[event].items():
+        for handler in self._listeners[event].items():
             try:
-                results.append(asyncio.get_event_loop().run_until_complete(callback(ev, name, **kwargs)))
+                results.append(asyncio.get_event_loop().run_until_complete(handler.handle(ev, **kwargs)))
             except Exception as e:
                 LOG.exception(e)
-                LOG.error(f"Failed to emit event '{event}' to '{name}'. Error message '{e!s}'.")
+                LOG.error(f"Failed to emit event '{event}' to '{handler.name}'. Error message '{e!s}'.")
 
         return results
 
-    def emit(self, event: str, data: any, **kwargs) -> Awaitable:
+    async def emit(self, event: str, data: any, **kwargs) -> Awaitable:
         """
         Emit an event.
 
@@ -216,18 +350,17 @@ class EventBus(metaclass=Singleton):
 
         """
         if event not in self._listeners:
-            return None
+            return []
 
         ev = Event(event=event, data=data)
-
+        LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data})
-        LOG.debug(f"Emitting event '{ev}'.")
 
         tasks = []
-        for name, callback in self._listeners[event].items():
+        for handler in self._listeners[event].values():
             try:
-                tasks.append(asyncio.create_task(callback(ev, name, **kwargs)))
+                tasks.append(handler.handle(ev, **kwargs))
             except Exception as e:
                 LOG.exception(e)
-                LOG.error(f"Failed to emit event '{event}' to '{name}'. Error message '{e!s}'.")
+                LOG.error(f"Failed to emit event '{ev.event}' to '{handler.name}'. Error message '{e!s}'.")
 
         return asyncio.gather(*tasks)

app/library/HttpAPI.py

@@ -24,9 +24,8 @@ from .cache import Cache
 from .common import Common
 from .config import Config
 from .DownloadQueue import DownloadQueue
-from .Emitter import Emitter
 from .encoder import Encoder
-from .Events import Events
+from .Events import EventBus, Events, message
 from .ffprobe import ffprobe
 from .M3u8 import M3u8
 from .Notifications import Notification, NotificationEvents
@@ -70,14 +69,13 @@ class HttpAPI(Common):
     def __init__(
         self,
         queue: DownloadQueue | None = None,
-        emitter: Emitter | None = None,
         encoder: Encoder | None = None,
         config: Config | None = None,
     ):
         self.queue = queue or DownloadQueue.get_instance()
-        self.emitter = emitter or Emitter.get_instance()
         self.encoder = encoder or Encoder()
         self.config = config or Config.get_instance()
+        self._notify = EventBus.get_instance()
 
         self.rootPath = str(Path(__file__).parent.parent.parent)
         self.routes = web.RouteTableDef()
@@ -793,7 +791,7 @@ class HttpAPI(Common):
                 status=web.HTTPInternalServerError.status_code,
             )
 
-        await self.emitter.emit(Events.PRESETS_UPDATE, presets)
+        await self._notify.emit(Events.PRESETS_UPDATE, data=presets)
         return web.json_response(data=presets, status=web.HTTPOk.status_code, dumps=self.encoder.encode)
 
     @route("GET", "api/tasks")
@@ -951,7 +949,7 @@ class HttpAPI(Common):
 
         if updated:
             self.queue.done.put(item)
-            await self.emitter.emit(Events.UPDATE, item.info)
+            await self._notify.emit(Events.UPDATE, data=item.info)
 
         return web.json_response(
             data=item.info,
@@ -1621,7 +1619,8 @@ class HttpAPI(Common):
             Response: The response object.
 
         """
-        data = {"type": "test", "message": "This is a test notification."}
+        data = message("test", "This is a test notification.")
-        await self.emitter.emit(Events.TEST, data)
+
+        await self._notify.emit(Events.TEST, data=data)
 
         return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=self.encoder.encode)

app/library/HttpSocket.py

@@ -7,7 +7,6 @@ import pty
 import shlex
 import time
 from datetime import UTC, datetime
-from typing import Any
 
 import anyio
 import socketio
@@ -16,9 +15,8 @@ from aiohttp import web
 from .common import Common
 from .config import Config
 from .DownloadQueue import DownloadQueue
-from .Emitter import Emitter
 from .encoder import Encoder
-from .Events import EventBus, Events
+from .Events import Event, EventBus, Events, error
 from .Presets import Presets
 from .Utils import arg_converter, is_downloaded
 
@@ -33,26 +31,25 @@ class HttpSocket(Common):
     config: Config
     sio: socketio.AsyncServer
     queue: DownloadQueue
-    emitter: Emitter
 
     def __init__(
         self,
         queue: DownloadQueue | None = None,
-        emitter: Emitter | None = None,
         encoder: Encoder | None = None,
         config: Config | None = None,
         sio: socketio.AsyncServer | None = None,
     ):
         self.config = config or Config.get_instance()
         self.queue = queue or DownloadQueue.get_instance()
-        self.emitter = emitter or Emitter.get_instance()
+        self._notify = EventBus.get_instance()
+
         self.sio = sio or socketio.AsyncServer(cors_allowed_origins="*")
         encoder = encoder or Encoder()
 
-        def emit(event: str, data: Any, **kwargs):
+        def emit(e: Event, _, **kwargs):
-            return self.sio.emit(event=event, data=encoder.encode(data), **kwargs)
+            return self.sio.emit(event=e.event, data=encoder.encode(e.data), **kwargs)
 
-        self.emitter.add_emitter([emit], local=False)
+        self._notify.subscribe("frontend", emit, f"{__class__.__name__}.socket_api")
 
         super().__init__(queue=queue, encoder=encoder, config=config)
 
@@ -80,8 +77,8 @@ class HttpSocket(Common):
             if hasattr(method, "_ws_event") and self.sio:
                 self.sio.on(method._ws_event)(method)  # type: ignore
 
-        EventBus.get_instance().subscribe(
+        self._notify.subscribe(
-            Events.ADD_URL, lambda data, _: self.add(**data.data), "socket_add_url"
+            Events.ADD_URL, lambda data, _: self.add(**data.data), f"{__class__.__name__}.socket_add_url"
         )
 
         # register the shutdown event.
@@ -90,11 +87,11 @@ class HttpSocket(Common):
     @ws_event
     async def cli_post(self, sid: str, data):
         if not self.config.console_enabled:
-            await self.emitter.error("Console is disabled.", to=sid)
+            await self._notify.emit(Events.ERROR, data=error("Console is disabled."), to=sid)
             return
 
         if not data:
-            await self.emitter.emit(Events.CLI_CLOSE, {"exitcode": 0}, to=sid)
+            await self._notify.emit(Events.CLI_CLOSE, data={"exitcode": 0}, to=sid)
             return
 
         try:
@@ -145,18 +142,18 @@ class HttpSocket(Common):
                         # No more output
                         if buffer:
                             # Emit any remaining partial line
-                            await self.emitter.emit(
+                            await self._notify.emit(
                                 Events.CLI_OUTPUT,
-                                {"type": "stdout", "line": buffer.decode("utf-8", errors="replace")},
+                                data={"type": "stdout", "line": buffer.decode("utf-8", errors="replace")},
                                 to=sid,
                             )
                         break
                     buffer += chunk
                     *lines, buffer = buffer.split(b"\n")
                     for line in lines:
-                        await self.emitter.emit(
+                        await self._notify.emit(
                             Events.CLI_OUTPUT,
-                            {"type": "stdout", "line": line.decode("utf-8", errors="replace")},
+                            data={"type": "stdout", "line": line.decode("utf-8", errors="replace")},
                             to=sid,
                         )
                 try:
@@ -173,58 +170,58 @@ class HttpSocket(Common):
             # Ensure reading is done
             await read_task
 
-            await self.emitter.emit(Events.CLI_CLOSE, {"exitcode": returncode}, to=sid)
+            await self._notify.emit(Events.CLI_CLOSE, data={"exitcode": returncode}, to=sid)
         except Exception as e:
             LOG.error(f"CLI execute exception was thrown for client '{sid}'.")
             LOG.exception(e)
-            await self.emitter.emit(Events.CLI_OUTPUT, {"type": "stderr", "line": str(e)}, to=sid)
+            await self._notify.emit(Events.CLI_OUTPUT, data={"type": "stderr", "line": str(e)}, to=sid)
-            await self.emitter.emit(Events.CLI_CLOSE, {"exitcode": -1}, to=sid)
+            await self._notify.emit(Events.CLI_CLOSE, data={"exitcode": -1}, to=sid)
 
     @ws_event
     async def add_url(self, sid: str, data: dict):
         url: str | None = data.get("url")
 
         if not url:
-            await self.emitter.error("No URL provided.", data={"unlock": True}, to=sid)
+            await self._notify.emit(Events.ERROR, data=error("No URL provided.", data={"unlock": True}), to=sid)
             return
 
         try:
             item = self.format_item(data)
         except ValueError as e:
-            await self.emitter.error(str(e), to=sid)
+            await self._notify.emit(Events.ERROR, data=error(str(e)), to=sid)
             return
 
         status = await self.add(**item)
-        await self.emitter.emit(event=Events.STATUS, data=status, to=sid)
+        await self._notify.emit(Events.STATUS, data=status, to=sid)
 
     @ws_event
     async def item_cancel(self, sid: str, id: str):
         if not id:
-            await self.emitter.error("Invalid request.", to=sid)
+            await self._notify.emit(Events.ERROR, data=error("Invalid request."), to=sid)
             return
 
         status: dict[str, str] = {}
         status = await self.queue.cancel([id])
         status.update({"identifier": id})
 
-        await self.emitter.emit(event=Events.ITEM_CANCEL, data=status)
+        await self._notify.emit(Events.ITEM_CANCEL, data=status)
 
     @ws_event
     async def item_delete(self, sid: str, data: dict):
         if not data:
-            await self.emitter.error("Invalid request.", to=sid)
+            await self._notify.emit(Events.ERROR, data=error("Invalid request."), to=sid)
             return
 
         id: str | None = data.get("id")
         if not id:
-            await self.emitter.error("Invalid request.", to=sid)
+            await self._notify.emit(Events.ERROR, data=error("Invalid request."), to=sid)
             return
 
         status: dict[str, str] = {}
         status = await self.queue.clear([id], remove_file=bool(data.get("remove_file", False)))
         status.update({"identifier": id})
 
-        await self.emitter.emit(event=Events.ITEM_DELETE, data=status)
+        await self._notify.emit(Events.ITEM_DELETE, data=status)
 
     @ws_event
     async def archive_item(self, _: str, data: dict):
@@ -275,36 +272,36 @@ class HttpSocket(Common):
         downloadPath: str = self.config.download_path
         data["folders"] = [name for name in os.listdir(downloadPath) if os.path.isdir(os.path.join(downloadPath, name))]
 
-        await self.emitter.emit(event=Events.INITIAL_DATA, data=data, to=sid)
+        await self._notify.emit(Events.INITIAL_DATA, data=data, to=sid)
 
     @ws_event
     async def pause(self, *_, **__):
         self.queue.pause()
-        await self.emitter.emit(event=Events.PAUSED, data={"paused": True, "at": time.time()})
+        await self._notify.emit(Events.PAUSED, data={"paused": True, "at": time.time()})
 
     @ws_event
     async def resume(self, *_, **__):
         self.queue.resume()
-        await self.emitter.emit(event=Events.PAUSED, data={"paused": False, "at": time.time()})
+        await self._notify.emit(Events.PAUSED, data={"paused": False, "at": time.time()})
 
     @ws_event
     async def ytdlp_convert(self, sid: str, data: dict):
         if not isinstance(data, dict) or "args" not in data:
-            await self.emitter.error("Invalid request or no options were given.", to=sid)
+            await self._notify.emit(Events.ERROR, data=error("Invalid request or no options were given."), to=sid)
             return
 
         args: str | None = data.get("args")
 
         if not args:
-            await self.emitter.error("no options were given.", to=sid)
+            await self._notify.emit(Events.ERROR, data=error("no options were given."), to=sid)
             return
 
         try:
-            await self.emitter.emit(event=Events.YTDLP_CONVERT, data=arg_converter(args), to=sid)
+            await self._notify.emit(Events.YTDLP_CONVERT, data=arg_converter(args), to=sid)
         except Exception as e:
             err = str(e).strip()
             err = err.split("\n")[-1] if "\n" in err else err
             LOG.error(f"Failed to convert args. '{err}'.")
-            await self.emitter.error(f"Failed to convert options. '{err}'.", to=sid)
+            await self._notify.emit(Events.ERROR, data=error(f"Failed to convert options. '{err}'."), to=sid)
 
         return

app/library/Notifications.py

@@ -7,10 +7,11 @@ from datetime import UTC, datetime
 from typing import Any
 
 import httpx
+from aiohttp import web
 
 from .config import Config
 from .encoder import Encoder
-from .Events import Events
+from .Events import EventBus, Event, Events
 from .ItemDTO import ItemDTO
 from .Singleton import Singleton
 from .Utils import ag, validate_uuid
@@ -98,6 +99,13 @@ class NotificationEvents:
     def get_events() -> dict[str, str]:
         return {k: v for k, v in vars(NotificationEvents).items() if not k.startswith("__") and not callable(v)}
 
+    def events() -> list:
+        return [
+            getattr(NotificationEvents, ev)
+            for ev in dir(NotificationEvents)
+            if not ev.startswith("_") and not callable(getattr(NotificationEvents, ev))
+        ]
+
     @staticmethod
     def is_valid(event: str) -> bool:
         return event in NotificationEvents.get_events().values()
@@ -142,6 +150,17 @@ class Notification(metaclass=Singleton):
 
         return Notification._instance
 
+    def attach(self, _: web.Application):
+        """
+        Attach the class to the application.
+
+        Args:
+            _ (web.Application): The aiohttp application.
+
+        """
+        self.load()
+        EventBus.get_instance().subscribe(NotificationEvents.events(), self.emit, f"{__class__.__name__}.emit")
+
     def get_targets(self) -> list[Target]:
         """Get the list of notification targets."""
         return self._targets
@@ -211,6 +230,35 @@ class Notification(metaclass=Singleton):
 
         return self
 
+    def make_target(self, target: dict) -> Target:
+        """
+        Make a notification target from a dictionary.
+
+        Args:
+            target (dict): The target details.
+
+        Returns:
+            Target: The notification target.
+
+        """
+        return Target(
+            id=target.get("id"),
+            name=target.get("name"),
+            on=target.get("on", []),
+            request=TargetRequest(
+                type=target.get("request", {}).get("type", "json"),
+                method=target.get("request", {}).get("method", "POST"),
+                url=target.get("request", {}).get("url"),
+                headers=[
+                    TargetRequestHeader(
+                        key=str(h.get("key", "")).strip(),
+                        value=str(h.get("value", "")).strip(),
+                    )
+                    for h in target.get("request", {}).get("headers", [])
+                ],
+            ),
+        )
+
     @staticmethod
     def validate(target: Target | dict) -> bool:
         """
@@ -340,40 +388,11 @@ class Notification(metaclass=Singleton):
             LOG.error(f"Error sending Notification event '{event}' id '{itemId}' to '{target.name}'. '{e}'.")
             return {"url": target.request.url, "status": 500, "text": str(e)}
 
-    def make_target(self, target: dict) -> Target:
+    def emit(self, e: Event, _, **kwargs):  # noqa: ARG002
-        """
-        Make a notification target from a dictionary.
-
-        Args:
-            target (dict): The target details.
-
-        Returns:
-            Target: The notification target.
-
-        """
-        return Target(
-            id=target.get("id"),
-            name=target.get("name"),
-            on=target.get("on", []),
-            request=TargetRequest(
-                type=target.get("request", {}).get("type", "json"),
-                method=target.get("request", {}).get("method", "POST"),
-                url=target.get("request", {}).get("url"),
-                headers=[
-                    TargetRequestHeader(
-                        key=str(h.get("key", "")).strip(),
-                        value=str(h.get("value", "")).strip(),
-                    )
-                    for h in target.get("request", {}).get("headers", [])
-                ],
-            ),
-        )
-
-    def emit(self, event, data, **kwargs):  # noqa: ARG002
         if len(self._targets) < 1:
             return False
 
-        if not NotificationEvents.is_valid(event):
+        if not NotificationEvents.is_valid(e.event):
             return False
 
-        return self.send(event, data)
+        return self.send(e.event, e.data)

app/library/Presets.py

@@ -8,7 +8,6 @@ from typing import Any
 from aiohttp import web
 
 from .config import Config
-from .Emitter import Emitter
 from .encoder import Encoder
 from .Events import EventBus, Events
 from .Singleton import Singleton
@@ -67,13 +66,12 @@ class Presets(metaclass=Singleton):
 
     _default_presets: list[Preset] = []
 
-    def __init__(self, file: str | None = None, emitter: Emitter | None = None, config: Config | None = None):
+    def __init__(self, file: str | None = None, config: Config | None = None):
         Presets._instance = self
 
         config = config or Config.get_instance()
 
         self._file: str = file or os.path.join(config.config_path, "presets.json")
-        self._emitter: Emitter = emitter or Emitter.get_instance()
 
         if os.path.exists(self._file) and "600" != oct(os.stat(self._file).st_mode)[-3:]:
             try:
@@ -85,7 +83,7 @@ class Presets(metaclass=Singleton):
             self._default_presets = [Preset(**preset) for preset in json.load(f)]
 
         EventBus.get_instance().subscribe(
-            Events.PRESETS_ADD, lambda data, _: self.add(**data.data), f"{__class__}.save"
+            Events.PRESETS_ADD, lambda data, _: self.add(**data.data), f"{__class__.__name__}.save"
         )
 
     @staticmethod

app/library/Scheduler.py

@@ -27,7 +27,7 @@ class Scheduler(metaclass=Singleton):
         self._loop = loop or asyncio.get_event_loop()
 
         EventBus.get_instance().subscribe(
-            Events.SCHEDULE_ADD, lambda data, _: self.add(**data.data), f"{__class__}.add"
+            Events.SCHEDULE_ADD, lambda data, _: self.add(**data.data), f"{__class__.__name__}.add"
         )
 
     @staticmethod

app/library/Tasks.py

@@ -11,9 +11,8 @@ import httpx
 from aiohttp import web
 
 from .config import Config
-from .Emitter import Emitter
 from .encoder import Encoder
-from .Events import Event, EventBus, Events
+from .Events import EventBus, Events, error, info, success
 from .Scheduler import Scheduler
 from .Singleton import Singleton
 
@@ -56,7 +55,6 @@ class Tasks(metaclass=Singleton):
     def __init__(
         self,
         file: str | None = None,
-        emitter: Emitter | None = None,
         loop: asyncio.AbstractEventLoop | None = None,
         config: Config | None = None,
         encoder: Encoder | None = None,
@@ -73,8 +71,8 @@ class Tasks(metaclass=Singleton):
         self._client = client or httpx.AsyncClient()
         self._encoder = encoder or Encoder()
         self._loop = loop or asyncio.get_event_loop()
-        self._emitter = emitter or Emitter.get_instance()
         self._scheduler = scheduler or Scheduler.get_instance()
+        self._notify = EventBus.get_instance()
 
         if os.path.exists(self._file) and "600" != oct(os.stat(self._file).st_mode)[-3:]:
             try:
@@ -82,7 +80,7 @@ class Tasks(metaclass=Singleton):
             except Exception:
                 pass
 
-        EventBus.get_instance().subscribe(Events.TASKS_ADD, lambda data, _: self.add(**data.data), f"{__class__}.save")
+        self._notify.subscribe(Events.TASKS_ADD, lambda data, _: self.add(**data.data), f"{__class__.__name__}.save")
 
     @staticmethod
     def get_instance() -> "Tasks":
@@ -296,23 +294,22 @@ class Tasks(metaclass=Singleton):
             LOG.info(f"Task '{task.id}: {task.name}' dispatched at '{timeNow}'.")
 
             tasks = []
-            tasks.append(self._emitter.info(f"Task '{task.name}' dispatched at '{timeNow}'."))
             tasks.append(
-                self._emitter.emit(
+                self._notify.emit(Events.LOG_INFO, data=info(f"Task '{task.name}' dispatched at '{timeNow}'."))
-                    event=Events.ADD_URL,
+            )
-                    data=Event(
+            tasks.append(
-                        id=task.id,
+                self._notify.emit(
-                        data={
+                    Events.ADD_URL,
-                            "url": task.url,
+                    data={
-                            "preset": preset,
+                        "url": task.url,
-                            "folder": folder,
+                        "preset": preset,
-                            "cookies": cookies,
+                        "folder": folder,
-                            "config": config,
+                        "cookies": cookies,
-                            "template": template,
+                        "config": config,
-                        },
+                        "template": template,
-                    ),
+                    },
-                    local=True,
+                    id=task.id,
-                )
+                ),
             )
 
             await asyncio.wait_for(asyncio.gather(*tasks), timeout=None)
@@ -322,8 +319,13 @@ class Tasks(metaclass=Singleton):
             ended = time.time()
             LOG.info(f"Task '{task.id}: {task.name}' completed at '{timeNow}' took '{ended - started:.2f}' seconds.")
 
-            await self._emitter.success(f"Task '{task.name}' completed in '{ended - started:.2f}' seconds.")
+            await self._notify.emit(
+                Events.LOG_SUCCESS,
+                data=success(f"Task '{task.name}' completed in '{ended - started:.2f}' seconds."),
+            )
         except Exception as e:
             timeNow = datetime.now(UTC).isoformat()
             LOG.error(f"Task '{task.id}: {task.name}' has failed to execute at '{timeNow}'. '{e!s}'.")
-            await self._emitter.error(f"Task '{task.name}' failed to execute at '{timeNow}'. '{e!s}'.")
+            await self._notify.emit(
+                Events.ERROR, data=error(f"Task '{task.name}' failed to execute at '{timeNow}'. '{e!s}'.")
+            )

app/main.py

@@ -11,7 +11,6 @@ import magic
 from aiohttp import web
 from library.config import Config
 from library.DownloadQueue import DownloadQueue
-from library.Emitter import Emitter
 from library.Events import EventBus, Events
 from library.HttpAPI import HttpAPI
 from library.HttpSocket import HttpSocket
@@ -60,10 +59,6 @@ class Main:
         self._http = HttpAPI(queue=self._queue)
         self._socket = HttpSocket(queue=self._queue)
 
-        Emitter.get_instance().add_emitter([Notification().emit], local=False).add_emitter(
-            [EventBus().emit], local=True
-        )
-
         self._app.on_cleanup.append(_close_connection)
 
     def _check_folders(self):
@@ -94,7 +89,7 @@ class Main:
         """
         Start the application.
         """
-        EventBus.get_instance().emit(Events.STARTUP, data={"app": self._app})
+        EventBus.get_instance().sync_emit(Events.STARTUP, data={"app": self._app})
 
         self._socket.attach(self._app)
         self._http.attach(self._app)
@@ -102,6 +97,9 @@ class Main:
         Scheduler.get_instance().attach(self._app)
         Tasks.get_instance().attach(self._app)
         Presets.get_instance().attach(self._app)
+        Notification.get_instance().attach(self._app)
+
+        EventBus.get_instance().sync_emit(Events.LOADED, data={"app": self._app})
 
         def started(_):
             LOG.info("=" * 40)