"""ClientAppLoader — dynamic client application loader. Ported from net.i2p.router.startup.LoadClientAppsJob. Loads client applications from configuration at router startup. Each app entry specifies a module path, class name, optional args, and a startup delay. """ from __future__ import annotations import importlib import logging import time logger = logging.getLogger(__name__) class ClientAppLoader: """Loads client applications from config at router startup.""" def __init__(self, config: list[dict] | None = None) -> None: self._config = config or [] self._loaded: list = [] def load_apps(self) -> list: """Load and instantiate all configured client apps. Config format: [ { "module": "mypackage.mymodule", "class": "MyApp", "args": ["arg1", "arg2"], # optional "delay": 0.0, # optional, seconds }, ... ] Returns list of instantiated app objects. """ self._loaded = [] for entry in self._config: module_path = entry.get("module", "") class_name = entry.get("class", "") args = entry.get("args", []) delay = entry.get("delay", 0.0) if not module_path or not class_name: logger.warning("Skipping incomplete app config: %s", entry) continue app = self._load_app(module_path, class_name, args, delay) if app is not None: self._loaded.append(app) logger.info("Loaded %d client apps", len(self._loaded)) return list(self._loaded) def _load_app(self, module_path: str, class_name: str, args: list, delay: float) -> object | None: """Load a single client app.""" if delay > 0: time.sleep(delay) try: module = importlib.import_module(module_path) cls = getattr(module, class_name) return cls(*args) except (ImportError, AttributeError) as e: logger.error("Failed to load %s.%s: %s", module_path, class_name, e) return None except Exception as e: logger.error("Error instantiating %s.%s: %s", module_path, class_name, e) return None @property def loaded_apps(self) -> list: return list(self._loaded)