INNER CODE UNIT · Python
_reload_configurations_from_notification
yihong1120/Construction-Hazard-Detection · main.py:151
async def _reload_configurations_from_notification(self) -> None:
"""Coalesce notifications that arrive during one configuration read."""
while self._config_reload_requested:
self._config_reload_requested = False
await self.reload_configurations()
def _log_config_reload_failure(self, task: asyncio.Task[None]) -> None:
"""Log callback reload failures instead of losing the task error."""
if task.cancelled():
return
error = task.exception()
if error is not None:
self.logger.error(f"[config] Immediate reload failed: {error}")
async def _cancel_config_reload_task(self) -> None:
"""Cancel an in-flight callback reload before process shutdown."""
task = self._config_reload_task
self._config_reload_task = None