RTCConfigFileMonitor

def __init__(self, rtc_file: str, enabled: bool = True):
    self.enabled = enabled
    self.rtc_file = os.path.abspath(rtc_file)
    self.watch_dir = os.path.dirname(self.rtc_file) or "."
    self._loop = asyncio.get_running_loop()
    self.on_rtc_config: Callable[[List[str], List[str], bytes], Any] = lambda stun_servers, turn_servers, rtc_config: logger_rtcice.warning("unhandled on_rtc_config")

    self.observer = Observer()
    self.observer.schedule(self, self.watch_dir, recursive=False)

async def start(self) -> None:
    """Starts the watchdog observer thread; no-op when disabled."""
    if not self.enabled:
        return

    await asyncio.to_thread(self.observer.start)
    logger_rtcice.info(f"RTC config file monitor started for: {self.rtc_file}")

def _shutdown_observer(self) -> None:
    """Stops the observer and joins its thread; runs off the event loop."""
    if self.observer.is_alive():
        self.observer.stop()
        self.observer.join()

async def stop(self) -> None:
    """Stops the watchdog observer; no-op when disabled."""
    if not self.enabled:
        return

    await asyncio.to_thread(self._shutdown_observer)
    logger_rtcice.info("RTC config file monitor stopped")

def _reload_config(self, src_path: str) -> None:
    """Reads, parses, and dispatches the updated RTC config.

    Runs on the watchdog thread; the callback dispatch is handed to the
    event loop via `call_soon_threadsafe`. The file is re-checked for
    trusted ownership/permissions on every reload because it can be
    replaced between events.
    """
    try:
        logger_rtcice.info(f"Detected RTC JSON file change: {src_path}")
        if not _is_trusted_config_file(self.rtc_file):
            logger_rtcice.error(
                f"Refusing to reload RTC config file '{self.rtc_file}': unsafe ownership or permissions."
            )
            return
        with open(self.rtc_file, 'rb') as f:
            data = f.read()

        stun_servers, turn_servers, rtc_config = parse_rtc_config(data)
        self._loop.call_soon_threadsafe(
            _schedule_rtc_callback,
            self._loop,
            self.on_rtc_config,
            stun_servers,
            turn_servers,
            rtc_config
        )
    except Exception as e:
        logger_rtcice.warning(f"Could not read or parse RTC JSON file: {self.rtc_file}: {e}")

def on_closed(self, event: Any) -> None:
    """Reloads after an in-place write of the config file."""
    if not isinstance(event, FileClosedEvent):
        return
    if os.path.abspath(event.src_path) != self.rtc_file:
        return
    self._reload_config(event.src_path)

def on_moved(self, event: Any) -> None:
    """Reloads when a temp file is renamed onto the config file."""
    dest = getattr(event, "dest_path", None)
    if dest and os.path.abspath(dest) == self.rtc_file:
        self._reload_config(dest)

def on_created(self, event: Any) -> None:
    """Reloads when the config file is created anew."""
    if os.path.abspath(event.src_path) == self.rtc_file:
        self._reload_config(event.src_path)