minor changes to the event bus dispatcher

app/library/DownloadQueue.py

@@ -108,15 +108,17 @@ class DownloadQueue(metaclass=Singleton):
 
         return DownloadQueue._instance
 
-    def attach(self, app: web.Application):
+    def attach(self, _: web.Application):
         """
         Attach the download queue to the application.
 
         Args:
-            app (web.Application): The application to attach the download queue to.
+            _ (web.Application): The application to attach the download queue to.
 
         """
-        app.on_startup.append(lambda _: self.initialize())
+        self._notify.subscribe(
+            Events.STARTED, lambda _, __: self.initialize(), f"{__class__.__name__}.{__class__.__name__}.initialize"
+        )
 
         Scheduler.get_instance().add(
             timer="* * * * *",

app/library/Events.py

@@ -1,6 +1,7 @@
 import asyncio
 import datetime
 import logging
+from typing import Any
 import uuid
 from collections.abc import Awaitable
 from dataclasses import dataclass, field
@@ -93,6 +94,7 @@ class Events:
 
     STARTUP = "startup"
     LOADED = "loaded"
+    STARTED = "started"
     SHUTDOWN = "shutdown"
 
     ADDED = "added"
@@ -346,44 +348,71 @@ class EventBus(metaclass=Singleton):
 
         return self
 
-    def sync_emit(self, event: str, data: any, **kwargs) -> list:
+    def sync_emit(self, event: str, data: Any, loop=None, wait: bool = True, **kwargs):
         """
-        Emit an event synchronously.
+        Emit event and (optionally) wait for results.
 
         Args:
             event (str): The event to emit.
-            data (any): The data to pass to the event.
+            data (Any): The data to pass to the event.
-            **kwargs: The keyword arguments to pass to the event.
+            loop (asyncio.AbstractEventLoop | None): The event loop to use. If None, the current running loop is used.
+            wait (bool): Whether to wait for the results of the event handlers. Defaults to True.
+            **kwargs: Additional 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 wait is False, a list of asyncio.Task objects is returned instead if we are in a running event loop,
+                  or the result of the coroutine if we are not in a running event loop.
         """
         if event not in self._listeners:
             return []
 
-        ev = Event(event=event, data=data)
+        async def emit_all():
-        LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data})
+            ev = Event(event=event, data=data)
+            LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data})
 
-        results = []
+            res: list = []
 
-        for handler in self._listeners[event].values():
+            for h in self._listeners[event].values():
-            try:
+                try:
-                results.append(asyncio.get_event_loop().run_until_complete(handler.handle(ev, **kwargs)))
+                    res.append(await h.handle(ev, **kwargs))
-            except Exception as e:
+                except Exception as e:
-                LOG.exception(e)
+                    LOG.exception(e)
-                LOG.error(f"Failed to emit event '{event}' to '{handler.name}'. Error message '{e!s}'.")
+                    LOG.error(f"Failed to emit event '{event}' to '{h.name}'. Error message '{e!s}'.")
 
-        return results
+            return res
 
-    async def emit(self, event: str, data: any, **kwargs) -> Awaitable:
+        try:
+            loop = loop or asyncio.get_running_loop()
+            in_same_loop: bool = asyncio.get_running_loop() is loop
+        except RuntimeError:
+            loop = None
+            in_same_loop = False
+
+        if loop is None or not loop.is_running():
+            return asyncio.run(emit_all())
+
+        if in_same_loop:
+            if wait:
+                msg = (
+                    "Calling EventsBus.sync_emit(...,wait=True) from within the running event loop would cause dead-lock. "
+                    "Use `await EventsBus.emit(...)` or `EventsBus.sync_emit(..., wait=False)`."
+                )
+                raise RuntimeError(msg)
+
+            return loop.create_task(emit_all())
+
+        fut = asyncio.run_coroutine_threadsafe(emit_all(), loop)
+        return fut.result() if wait else fut
+
+    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.
+            data (Any): The data to pass to the event.
             **kwargs: The keyword arguments to pass to the event.
 
         Returns:

app/library/Scheduler.py

@@ -74,9 +74,31 @@ class Scheduler(metaclass=Singleton):
         return self._jobs
 
     def get(self, id: str) -> Cron | None:
-        """Return the job by id."""
+        """
+        Get a job by its id.
+
+        Args:
+            id (str): The id of the job.
+
+        Returns:
+            Cron | None: The job if it exists, None otherwise
+
+        """
         return self._jobs.get(id)
 
+    def has(self, id: str) -> bool:
+        """
+        Check if a job with the given id exists.
+
+        Args:
+            id (str): The id of the job.
+
+        Returns:
+            bool: True if the job exists, False otherwise
+
+        """
+        return id in self._jobs
+
     def add(
         self, timer: str, func: callable, args: tuple = (), kwargs: dict | None = None, id: str | None = None
     ) -> str:

app/library/Tasks.py

@@ -162,11 +162,12 @@ class Tasks(metaclass=Singleton):
                     from cronsim import CronSim
 
                     cs = CronSim(task.timer, datetime.now(UTC))
-                    schedule_time = cs.explain()
+                    schedule_time: str = cs.explain()
                 except Exception:
                     schedule_time = task.timer
 
-                LOG.info(f"Task '{task.name}' queued to be executed '{schedule_time}'.")
+                if not has_tasks:
+                    LOG.info(f"Task '{task.name}' queued to be executed '{schedule_time}'.")
             except Exception as e:
                 LOG.exception(e)
                 LOG.error(f"Failed to queue '{i}: {task.name}'. '{e!s}'.")
@@ -185,8 +186,11 @@ class Tasks(metaclass=Singleton):
             return self
 
         for task in self._tasks:
+            if not self._scheduler.has(task.id):
+                continue
+
             try:
-                LOG.info(f"Stopping '{task.name}'.")
+                LOG.debug(f"Stopping '{task.name}'.")
                 self._scheduler.remove(task.id)
             except Exception as e:
                 if not shutdown:

app/main.py

@@ -38,13 +38,14 @@ class Main:
     def __init__(self, is_native: bool = False):
         self._config = Config.get_instance(is_native=is_native)
         self._app = web.Application()
+        self.loop = asyncio.get_event_loop()
+        self._app.on_shutdown.append(self.on_shutdown)
 
         Services.get_instance().add("app", self._app)
 
         if self._config.debug:
-            loop = asyncio.get_event_loop()
+            self.loop.set_debug(True)
-            loop.set_debug(True)
+            self.loop.slow_callback_duration = 0.05
-            loop.slow_callback_duration = 0.05
 
         self._check_folders()
 
@@ -97,6 +98,9 @@ class Main:
             LOG.error(f"Could not create database file at '{self._config.db_file}'. {e!s}")
             raise
 
+    async def on_shutdown(self, _: web.Application):
+        await EventBus.get_instance().emit(Events.SHUTDOWN, data={"app": self._app})
+
     def start(self, host: str | None = None, port: int | None = None, cb=None):
         """
         Start the application.
@@ -122,6 +126,9 @@ class Main:
             LOG.info("=" * 40)
             LOG.info(f"YTPTube {self._config.version} - started on http://{host}:{port}{self._config.base_path}")
             LOG.info("=" * 40)
+
+            EventBus.get_instance().sync_emit(Events.STARTED, data={"app": self._app}, loop=self.loop, wait=False)
+
             if cb:
                 cb()
 
@@ -138,7 +145,7 @@ class Main:
             host=host,
             port=port,
             reuse_port="win32" != sys.platform,
-            loop=asyncio.get_event_loop(),
+            loop=self.loop,
             access_log=HTTP_LOGGER,
             print=started,
             handle_signals=cb is None,