From f6e3c944327853ae751890291b6c2fee73f37bc8 Mon Sep 17 00:00:00 2001 From: arabcoders Date: Mon, 23 Jun 2025 01:00:36 +0300 Subject: [PATCH] minor changes to the event bus dispatcher --- app/library/DownloadQueue.py | 8 +++-- app/library/Events.py | 63 ++++++++++++++++++++++++++---------- app/library/Scheduler.py | 24 +++++++++++++- app/library/Tasks.py | 10 ++++-- app/main.py | 15 ++++++--- 5 files changed, 92 insertions(+), 28 deletions(-) diff --git a/app/library/DownloadQueue.py b/app/library/DownloadQueue.py index 89a6b535..e37703a5 100644 --- a/app/library/DownloadQueue.py +++ b/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="* * * * *", diff --git a/app/library/Events.py b/app/library/Events.py index 937b743e..74c2f240 100644 --- a/app/library/Events.py +++ b/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. - **kwargs: The keyword arguments to pass to the event. + data (Any): The data 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) - LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data}) + async def emit_all(): + 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(): - 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}'.") + for h in self._listeners[event].values(): + try: + res.append(await h.handle(ev, **kwargs)) + except Exception as e: + LOG.exception(e) + 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: diff --git a/app/library/Scheduler.py b/app/library/Scheduler.py index e26d49fe..4b03b509 100644 --- a/app/library/Scheduler.py +++ b/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: diff --git a/app/library/Tasks.py b/app/library/Tasks.py index 67e0973c..9355daa7 100644 --- a/app/library/Tasks.py +++ b/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: diff --git a/app/main.py b/app/main.py index ee7fcfd5..4693b7f4 100644 --- a/app/main.py +++ b/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() - loop.set_debug(True) - loop.slow_callback_duration = 0.05 + self.loop.set_debug(True) + self.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,