Merge pull request #221 from arabcoders/dev

Switiching to EventBus & updated yt-dlp to 2025.3.21

Pipfile.lock

@@ -1299,12 +1299,12 @@
         },
         "yt-dlp": {
             "hashes": [
-                "sha256:3ed218eaeece55e9d715afd41abc450dc406ee63bf79355169dfde312d38fdb8",
+                "sha256:5bcf47b2897254ea3816935a8dde47d243bff556782cced6b16a2b85e6b682ba",
-                "sha256:f33ca76df2e4db31880f2fe408d44f5058d9f135015b13e50610dfbe78245bea"
+                "sha256:80d5ce15f9223e0c27020b861a4c5b72c6ba5d6c957c1b8fd2a022a69783f482"
             ],
             "index": "pypi",
             "markers": "python_version >= '3.9'",
-            "version": "==2025.2.19"
+            "version": "==2025.3.21"
         }
     },
     "develop": {}

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 .EventsSubscriber 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, data=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 .EventsSubscriber 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

@@ -0,0 +1,364 @@
+import asyncio
+import datetime
+import logging
+import uuid
+from collections.abc import Awaitable
+from dataclasses import dataclass, field
+
+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"
+    UPDATED = "updated"
+    COMPLETED = "completed"
+    CANCELLED = "cancelled"
+    CLEARED = "cleared"
+    ERROR = "error"
+    LOG_INFO = "log_info"
+    LOG_SUCCESS = "log_success"
+
+    INITIAL_DATA = "initial_data"
+    YTDLP_CONVERT = "ytdlp_convert"
+    ITEM_DELETE = "item_delete"
+    ITEM_CANCEL = "item_cancel"
+    STATUS = "status"
+    CLI_CLOSE = "cli_close"
+    CLI_OUTPUT = "cli_output"
+    UPDATE = "update"
+    TEST = "test"
+    ADD_URL = "add_url"
+
+    CLI_POST = "cli_post"
+    PAUSED = "paused"
+
+    TASKS_ADD = "task_add"
+    TASK_DISPATCHED = "task_dispatched"
+    TASK_FINISHED = "task_finished"
+    TASK_ERROR = "task_error"
+
+    PRESETS_ADD = "presets_add"
+    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.INITIAL_DATA,
+            Events.ADDED,
+            Events.ERROR,
+            Events.LOG_INFO,
+            Events.LOG_SUCCESS,
+            Events.COMPLETED,
+            Events.CANCELLED,
+            Events.CLEARED,
+            Events.UPDATED,
+            Events.UPDATE,
+            Events.PAUSED,
+            Events.PRESETS_UPDATE,
+            Events.STATUS,
+            Events.CLI_CLOSE,
+            Events.CLI_OUTPUT,
+        ]
+
+
+@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(uuid.uuid4()), init=False)
+    """The id of the event."""
+
+    created_at: 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_at": self.created_at, "event": self.event, "data": self.data}
+
+    def __repr__(self):
+        return f"Event(id={self.id}, created_at={self.created_at}, 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_at={self.created_at}, 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.
+    """
+
+    _instance = None
+    """the instance of the EventsSubscriber"""
+
+    _listeners: dict[str, list[str, EventListener]] = {}
+    """The listeners for the events."""
+
+    def __init__(self):
+        EventBus._instance = self
+
+    @staticmethod
+    def get_instance() -> "EventBus":
+        """
+        Get the instance of the EventsSubscriber.
+
+        Returns:
+            EventsSubscriber: The instance of the EventsSubscriber
+
+        """
+        if not EventBus._instance:
+            EventBus._instance = EventBus()
+
+        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.
+            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(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] = EventListener(name, callback)
+
+        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]
+
+        for e in event:
+            if e in self._listeners and name in self._listeners[e]:
+                del self._listeners[e][name]
+
+        return self
+
+    def sync_emit(self, event: str, data: any, **kwargs) -> list:
+        """
+        Emit an event synchronously.
+
+        Args:
+            event (str): The event to emit.
+            data (any): The data to pass to the event.
+            **kwargs: The keyword arguments to pass to the event.
+
+        Returns:
+            list: The results are the return values of the coroutines. If the coroutine raises an exception,
+                  the exception is caught and logged. If event does not exist, an empty list is returned.
+
+        """
+        if event not in self._listeners:
+            return []
+
+        ev = Event(event=event, data=data)
+        LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data})
+
+        results = []
+        for handler in self._listeners[event].items():
+            try:
+                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 '{handler.name}'. Error message '{e!s}'.")
+
+        return results
+
+    async def emit(self, event: str, data: any, **kwargs) -> Awaitable:
+        """
+        Emit an event.
+
+        Args:
+            event (str): The event to emit.
+            data (any): The data to pass to the event.
+            **kwargs: The keyword arguments to pass to the event.
+
+        Returns:
+            Awaitable: The task that was created to run the event.
+
+        """
+        if event not in self._listeners:
+            return []
+
+        ev = Event(event=event, data=data)
+        LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data})
+
+        tasks = []
+        for handler in self._listeners[event].values():
+            try:
+                tasks.append(handler.handle(ev, **kwargs))
+            except Exception as e:
+                LOG.exception(e)
+                LOG.error(f"Failed to emit event '{ev.event}' to '{handler.name}'. Error message '{e!s}'.")
+
+        return asyncio.gather(*tasks)

app/library/EventsSubscriber.py

@@ -1,193 +0,0 @@
-import asyncio
-import logging
-from collections.abc import Awaitable
-from dataclasses import dataclass
-
-from .Singleton import Singleton
-
-LOG = logging.getLogger("EventsSubscriber")
-
-
-class Events:
-    """
-    The events that can be emitted.
-    """
-
-    STARTUP = "startup"
-    SHUTDOWN = "shutdown"
-
-    ADDED = "added"
-    UPDATED = "updated"
-    COMPLETED = "completed"
-    CANCELLED = "cancelled"
-    CLEARED = "cleared"
-    ERROR = "error"
-    LOG_INFO = "log_info"
-    LOG_SUCCESS = "log_success"
-
-    INITIAL_DATA = "initial_data"
-    YTDLP_CONVERT = "ytdlp_convert"
-    ITEM_DELETE = "item_delete"
-    ITEM_CANCEL = "item_cancel"
-    STATUS = "status"
-    CLI_CLOSE = "cli_close"
-    CLI_OUTPUT = "cli_output"
-    UPDATE = "update"
-    TEST = "test"
-    ADD_URL = "add_url"
-
-    CLI_POST = "cli_post"
-    PAUSED = "paused"
-
-    TASKS_ADD = "task_add"
-    TASK_DISPATCHED = "task_dispatched"
-    TASK_FINISHED = "task_finished"
-    TASK_ERROR = "task_error"
-
-    PRESETS_ADD = "presets_add"
-    PRESETS_UPDATE = "presets_update"
-    SCHEDULE_ADD = "schedule_add"
-
-
-@dataclass(kw_only=True)
-class Event:
-    id: str
-    data: dict
-
-
-class EventsSubscriber(metaclass=Singleton):
-    """
-    This class is used to subscribe to and emit events to the registered listeners.
-    """
-
-    _instance = None
-    """the instance of the EventsSubscriber"""
-
-    _listeners: dict[str, list[str, Awaitable]] = {}
-    """The listeners for the events."""
-
-    def __init__(self):
-        EventsSubscriber._instance = self
-
-    @staticmethod
-    def get_instance() -> "EventsSubscriber":
-        """
-        Get the instance of the EventsSubscriber.
-
-        Returns:
-            EventsSubscriber: The instance of the EventsSubscriber
-
-        """
-        if not EventsSubscriber._instance:
-            EventsSubscriber._instance = EventsSubscriber()
-        return EventsSubscriber._instance
-
-    def subscribe(self, event: str | list | tuple, id: str, callback: Awaitable) -> "EventsSubscriber":
-        """
-        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.
-
-        Returns:
-            EventsSubscriber: The instance of the EventsSubscriber
-
-        """
-        if isinstance(event, str):
-            event = [event]
-
-        for e in event:
-            if e not in self._listeners:
-                self._listeners[e] = {}
-
-            self._listeners[e][id] = callback
-
-        return self
-
-    def unsubscribe(self, event: str | list | tuple, id: str) -> "EventsSubscriber":
-        """
-        Unsubscribe from an event.
-
-        Args:
-            event (str): The event to unsubscribe from.
-            id (str): The id of the subscriber.
-
-        Returns:
-            EventsSubscriber: The instance of the EventsSubscriber
-
-        """
-        if isinstance(event, str):
-            event = [event]
-
-        for e in event:
-            if e in self._listeners and id in self._listeners[e]:
-                del self._listeners[e][id]
-
-        return self
-
-    def emit_sync(self, event: str, *args, **kwargs):
-        """
-        Emit an event synchronously.
-
-        Args:
-            event (str): The event to emit.
-            *args: The arguments to pass to the event.
-            **kwargs: The keyword arguments to pass to the event.
-
-        Returns:
-            list: The results are the return values of the coroutines. If the coroutine raises an exception,
-                  the exception is caught and logged. If event does not exist, an empty list is returned.
-
-        """
-        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"]
-
-                results.append(asyncio.get_event_loop().run_until_complete(callback(event, data)))
-            except Exception as e:
-                LOG.exception(e)
-                LOG.error(f"Failed to emit event '{event}' to '{id}'. Error message '{e!s}'.")
-
-        return results
-
-    def emit(self, event: str, *args, **kwargs):
-        """
-        Emit an event.
-
-        Args:
-            event (str): The event to emit.
-            *args: The arguments to pass to the event.
-            **kwargs: The keyword arguments to pass to the event.
-
-        Returns:
-            Awaitable: The task that was created to run the event.
-
-        """
-        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})
-
-                tasks.append(asyncio.create_task(callback(event, data)))
-            except Exception as e:
-                LOG.exception(e)
-                LOG.error(f"Failed to emit event '{event}' to '{id}'. 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 .EventsSubscriber 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 .EventsSubscriber import Event, Events, EventsSubscriber
+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,13 +77,11 @@ 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)
+        self._notify.subscribe(
-
+            Events.ADD_URL,
-        async def handle_event(_: str, data: Event):
+            lambda data, _, **kwargs: self.add(**data.data),  # noqa: ARG005
-            LOG.debug(f"Event received. '{data}'")
+            f"{__class__.__name__}.socket_add_url",
-            await self.add(**data.data)
+        )
-
-        EventsSubscriber.get_instance().subscribe(Events.ADD_URL, "socket_add_url", handle_event)
 
         # register the shutdown event.
         app.on_shutdown.append(self.on_shutdown)
@@ -94,11 +89,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:
@@ -149,18 +144,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:
@@ -177,58 +172,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):
@@ -279,36 +274,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 .EventsSubscriber 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
@@ -204,13 +223,42 @@ class Notification(metaclass=Singleton):
                 self._targets.append(target)
 
                 LOG.info(
-                    f"Will send '{target.on if len(target.on) > 0 else 'all'}' as {target.request.type} notification events to '{target.name}'."
+                    f"Will send {target.request.type} request on '{', '.join(target.on) if len(target.on) > 0 else 'all events'}' to '{target.name}'."
                 )
             except Exception as e:
                 LOG.error(f"Error loading notification target '{target}'. '{e!s}'")
 
         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:
         """
@@ -275,37 +323,35 @@ class Notification(metaclass=Singleton):
 
         return True
 
-    async def send(self, event: str, item: ItemDTO | dict) -> list[dict]:
+    async def send(self, ev: Event) -> list[dict]:
         if len(self._targets) < 1:
             return []
 
-        if not isinstance(item, ItemDTO) and not isinstance(item, dict):
+        if not isinstance(ev.data, ItemDTO) and not isinstance(ev.data, dict):
-            LOG.debug(f"Received invalid item type '{type(item)}' with event '{event}'.")
+            LOG.debug(f"Received invalid item type '{type(ev.data)}' with event '{ev.event}'.")
             return []
 
         tasks = []
 
         for target in self._targets:
-            if len(target.on) > 0 and event not in target.on and "test" != event:
+            if len(target.on) > 0 and ev.event not in target.on and "test" != ev.event:
                 continue
 
-            tasks.append(self._send(event, target, item))
+            tasks.append(self._send(target, ev))
 
         return await asyncio.gather(*tasks)
 
-    async def _send(self, event: str, target: Target, item: ItemDTO | dict) -> dict:
+    async def _send(self, target: Target, ev: Event) -> dict:
         try:
-            itemId = item.get("id", item.get("_id", "??"))
+            LOG.info(f"Sending Notification event '{ev.event}: {ev.id}' to '{target.name}'.")
-        except Exception:
-            itemId = "??"
 
-        try:
-            LOG.info(f"Sending Notification event '{event}' id '{itemId}' to '{target.name}'.")
             reqBody = {
                 "method": target.request.method.upper(),
                 "url": target.request.url,
                 "headers": {
                     "User-Agent": f"YTPTube/{APP_VERSION}",
+                    "X-Event-Id": ev.id,
+                    "X-Event": ev.event,
                     "Content-Type": "application/json"
                     if "json" == target.request.type.lower()
                     else "application/x-www-form-urlencoded",
@@ -316,20 +362,16 @@ class Notification(metaclass=Singleton):
                 for h in target.request.headers:
                     reqBody["headers"][h.key] = h.value
 
-            reqBody["json" if "json" == target.request.type.lower() else "data"] = {
+            reqBody["json" if "json" == target.request.type.lower() else "data"] = self._deep_unpack(ev.serialize())
-                "event": event,
-                "created_at": datetime.now(tz=UTC).isoformat(),
-                "payload": item.__dict__ if isinstance(item, ItemDTO) else item,
-            }
 
             if "form" == target.request.type.lower():
-                reqBody["data"]["payload"] = self._encoder.encode(reqBody["data"]["payload"])
+                reqBody["data"]["data"] = self._encoder.encode(reqBody["data"]["data"])
 
             response = await self._client.request(**reqBody)
 
             respData = {"url": target.request.url, "status": response.status_code, "text": response.text}
 
-            msg = f"Notification target '{target.name}' Responded to event '{event}' id '{itemId}' with status '{response.status_code}'."
+            msg = f"Notification target '{target.name}' Responded to event '{ev.event}: {ev.id}' with status '{response.status_code}'."
             if self._debug and respData.get("text"):
                 msg += f" body '{respData.get('text','??')}'."
 
@@ -337,43 +379,28 @@ class Notification(metaclass=Singleton):
 
             return respData
         except Exception as e:
-            LOG.error(f"Error sending Notification event '{event}' id '{itemId}' to '{target.name}'. '{e}'.")
+            LOG.exception(e)
-            return {"url": target.request.url, "status": 500, "text": str(e)}
+            LOG.error(f"Error sending Notification event '{ev.event}: {ev.id}' to '{target.name}'. '{e!s}'.")
+            return {"url": target.request.url, "status": 500, "text": str(ev)}
 
-    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
+            return []
 
-        if not NotificationEvents.is_valid(event):
+        if not NotificationEvents.is_valid(e.event):
-            return False
+            return []
 
-        return self.send(event, data)
+        return self.send(e)
+
+    def _deep_unpack(self, data: dict) -> dict:
+        for k, v in data.items():
+            if isinstance(v, dict):
+                data[k] = self._deep_unpack(v)
+            if isinstance(v, list):
+                data[k] = [self._deep_unpack(i) for i in v]
+            if isinstance(v, datetime):
+                data[k] = v.isoformat()
+            if isinstance(v, ItemDTO):
+                data[k] = v.serialize()
+
+        return data

app/library/Presets.py

@@ -8,9 +8,8 @@ from typing import Any
 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")
@@ -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:
@@ -81,13 +79,14 @@ 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, _, **kwargs: self.add(**data.data),  # noqa: ARG005
+            f"{__class__.__name__}.save",
+        )
 
     @staticmethod
     def get_instance() -> "Presets":

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,11 @@ class Scheduler(metaclass=Singleton):
 
         self._loop = loop or asyncio.get_event_loop()
 
-        def handle_event(_, e: Event):
+        EventBus.get_instance().subscribe(
-            self.add(**e.data)
+            Events.SCHEDULE_ADD,
-
+            lambda data, _, **kwargs: self.add(**data.data),  # noqa: ARG005
-        EventsSubscriber.get_instance().subscribe(Events.SCHEDULE_ADD, f"{__class__}.add", handle_event)
+            f"{__class__.__name__}.add",
+        )
 
     @staticmethod
     def get_instance() -> "Scheduler":
@@ -76,8 +77,9 @@ class Scheduler(metaclass=Singleton):
         """Return the job by id."""
         return self._jobs.get(id)
 
-    def add(self, timer: str, func: callable, args: tuple = (),
+    def add(
-            kwargs: dict | None = None, id: str | None = None) -> str:
+        self, timer: str, func: callable, args: tuple = (), kwargs: dict | None = None, id: str | None = None
+    ) -> str:
         """
         Add a job to the schedule.
 

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 .EventsSubscriber import Event, Events, EventsSubscriber
+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,11 +80,6 @@ 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)
-
     @staticmethod
     def get_instance() -> "Tasks":
         """
@@ -113,6 +106,11 @@ class Tasks(metaclass=Singleton):
 
         """
         self.load()
+        self._notify.subscribe(
+            Events.TASKS_ADD,
+            lambda data, _, **kwargs: self.add(**data.data),  # noqa: ARG005
+            f"{__class__.__name__}.save",
+        )
 
     def get_all(self) -> list[Task]:
         """Return the tasks."""
@@ -299,23 +297,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)
@@ -325,8 +322,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/library/encoder.py

@@ -5,6 +5,8 @@ from pathlib import Path
 from yt_dlp.networking.impersonate import ImpersonateTarget
 from yt_dlp.utils import DateRange
 
+from .ItemDTO import ItemDTO
+
 
 class Encoder(json.JSONEncoder):
     """
@@ -26,6 +28,9 @@ class Encoder(json.JSONEncoder):
         if isinstance(o, ImpersonateTarget):
             return str(o)
 
+        if isinstance(o, ItemDTO):
+            return o.serialize()
+
         if isinstance(o, object):
             if hasattr(o, "serialize"):
                 return o.serialize()

app/main.py

@@ -11,8 +11,7 @@ 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.EventsSubscriber import Events, EventsSubscriber
 from library.HttpAPI import HttpAPI
 from library.HttpSocket import HttpSocket
 from library.Notifications import Notification
@@ -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(
-            [EventsSubscriber().emit], local=True
-        )
-
         self._app.on_cleanup.append(_close_connection)
 
     def _check_folders(self):
@@ -94,7 +89,7 @@ class Main:
         """
         Start the application.
         """
-        EventsSubscriber.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)

ui/components/History.vue

@@ -411,7 +411,7 @@ const setIcon = item => {
 }
 
 const setIconColor = item => {
-  if (item.status === 'finished') {
+  if ('finished' === item.status) {
     return 'has-text-success'
   }
 

ui/components/Queue.vue

@@ -76,11 +76,10 @@
               </div>
               <div class="column is-half-mobile has-text-centered is-text-overflow">
                 <span class="icon-text">
-                  <span class="icon" :class="{ 'has-text-success': item.status == 'downloading' }">
+                  <span class="icon" :class="setIconColor(item)">
-                    <i class="fas" :class="setIcon(item)" />
+                    <i class="fas fa-solid" :class="setIcon(item)" />
                   </span>
-                  <span v-if="item.status == 'downloading' && item.is_live">Live Streaming</span>
+                  <span v-text="setStatus(item)" />
-                  <span v-else>{{ ucFirst(item.status) }}</span>
                 </span>
               </div>
               <div class="column is-half-mobile has-text-centered is-text-overflow">
@@ -153,7 +152,7 @@
 
 <script setup>
 import moment from 'moment'
-import { useStorage } from '@vueuse/core'
+import { set, useStorage } from '@vueuse/core'
 import { ucFirst } from '~/utils/index'
 import { isEmbedable, getEmbedable } from '~/utils/embedable'
 
@@ -184,24 +183,62 @@ const hasQueuedItems = computed(() => stateStore.count('queue') > 0)
 
 const setIcon = item => {
   if ('downloading' === item.status && item.is_live) {
-    return 'fa-solid fa-globe';
+    return 'fa-globe fa-spin';
   }
 
   if ('downloading' === item.status) {
-    return 'fa-solid fa-circle-check';
+    return 'fa-download';
   }
 
   if ('postprocessing' === item.status) {
-    return 'fa-solid fa-cog fa-spin';
+    return 'fa-cog fa-spin';
   }
 
   if (null === item.status && true === config.paused) {
-    return 'fa-solid fa-pause-circle';
+    return 'fa-pause-circle';
   }
 
-  return 'fa-solid fa-spinner fa-spin';
+  if (!item.status) {
+    return 'fa-question';
+  }
+
+  return 'fa-spinner fa-spin';
 }
 
+const setStatus = item => {
+  if (null === item.status && true === config.paused) {
+    return 'Paused';
+  }
+
+  if ('downloading' === item.status && item.is_live) {
+    return 'Live Streaming';
+  }
+
+  if (!item.status) {
+    return 'Unknown..';
+  }
+
+  return ucFirst(item.status)
+}
+
+const setIconColor = item => {
+  if (item.status === 'downloading') {
+    return 'has-text-success'
+  }
+
+  if ('postprocessing' === item.status) {
+    return 'has-text-info'
+  }
+
+  if (null === item.status && true === config.paused) {
+    return 'has-text-warning'
+  }
+
+  return ''
+}
+
+
+
 const ETAPipe = value => {
   if (value === null || 0 === value) {
     return 'Live';

ui/pages/console.vue

@@ -103,6 +103,14 @@ const runCommand = async () => {
     return
   }
 
+  if (command.value.startsWith('yt-dlp')) {
+    command.value = command.value.replace(/^yt-dlp/, '').trim()
+    await nextTick()
+    if ('' === command.value) {
+      return
+    }
+  }
+
   if (!terminal.value) {
     terminal.value = new Terminal({
       fontSize: 14,

ui/pages/notifications.vue

@@ -124,6 +124,12 @@ div.is-centered {
             However this might not be enough to remove credentials from the exported data. it's your responsibility
             to ensure that the exported data does not contain any sensitive information for sharing.
           </li>
+          <li>
+            When you set the request type as <code>Form</code>, the event data will be JSON encoded and sent as the
+            and sent as <code>...&data=json_string</code>, only the <code>data</code> field will be JSON encoded. The
+            other keys <code>id</code>, <code>event</code> and <code>created_at</code> will be sent as they are.
+          </li>
+          <li>We also send two special headers <code>X-Event-ID</code> and <code>X-Event</code> with the request.</li>
         </ul>
       </Message>
     </div>

ui/stores/SocketStore.js

@@ -70,7 +70,7 @@ export const useSocketStore = defineStore('socket', () => {
         return
       }
 
-      toast.info(`Download cancelled: ${ag(stateStore.get('queue', id, {}), id)}`);
+      toast.warning(`Download cancelled: ${ag(stateStore.get('queue', id, {}), 'title')}`);
 
       if (true === stateStore.has('queue', id)) {
         stateStore.remove('queue', id);