import logging from defence360agent import files from defence360agent.contracts import config, messages from defence360agent.contracts.plugins import MessageSource from defence360agent.utils import recurring_check logger = logging.getLogger(__name__) class FilesRecurringUpdateTask(MessageSource): async def _on_files_update( self, index: files.Index, is_updated: bool ) -> None: if is_updated: message = messages.MessageType.FilesUpdated(index.type, index) await self._sink.process_message(message) async def create_source(self, loop, sink): self._loop = loop self._sink = sink self._task = loop.create_task(self._update_task()) # subscribe to file updates for type_ in files.Index.types(): files.Index.add_hook(type_, self._on_files_update) async def shutdown(self): self._task.cancel() # CancelledError is handled by @recurring_check(): await self._task @recurring_check(config.FilesUpdate.PERIOD) async def _update_task(self): await files.update_and_log_error()