from logging import getLogger from defence360agent.api import inactivity from defence360agent.contracts.config import SimpleRpc from defence360agent.contracts.plugins import MessageSink from defence360agent.utils import clip, fail_agent_service, recurring_check logger = getLogger(__name__) class IdleTimeOutCheck(MessageSink): async def create_sink(self, loop): self._loop = loop if SimpleRpc.SOCKET_ACTIVATION: inactivity.track.reset_timer() self._task = loop.create_task( recurring_check( period=clip( SimpleRpc.INACTIVITY_TIMEOUT // 5, low=1, high=60 ), )( self._check_timeout, )() ) else: self._task = None async def shutdown(self): if self._task: self._task.cancel() # CancelledError is handled by @recurring_check(): await self._task async def _check_timeout(self): logger.info("Periodical check %s ", inactivity.track) if inactivity.track.is_timeout(): logger.warning("Shutting down due to inactivity.") fail_agent_service()