Source code for asab.timer
import asyncio
[docs]class Timer(object):
'''
T.__init__(app, handler, autorestart=False) -> Timer.
The relative and optionally repeating timer for asyncio.
This class is simple relative timer that generate an event after a given time, and optionally repeating in regular intervals after that.
:param app: An ASAB application.
:param handler: A coro or future that will be called when a timer triggers.
:param boolean autorestart: If `True` then a timer will be automatically restarted after triggering.
:ivar Handler: A coro or future that will be called when a timer triggers.
:ivar Task: A future that represent the timer task.
:ivar App: An ASAB app.
:ivar boolean AutoRestart: If `True` then a timer will be automatically restarted after triggering.
The timer object is initialized as stopped.
*Note*: The implementation idea is borrowed from "`Python - Timer with asyncio/coroutine <https://stackoverflow.com/questions/45419723/python-timer-with-asyncio-coroutine>`_" question on StackOverflow.
'''
def __init__(self, app, handler, autorestart=False):
self.App = app
self.Handler = handler
self.Task = None
self.AutoRestart = autorestart
app.PubSub.subscribe("Application.stop!", self._on_stop)
[docs] def start(self, timeout):
'''
Start the timer.
:param float/int timeout: A timer delay in seconds.
'''
if self.is_started():
raise RuntimeError("Timer is already started")
self.Task = asyncio.ensure_future(self._job(timeout))
[docs] def stop(self):
'''
Stop the timer.
'''
if self.Task is not None:
self.Task.cancel()
self.Task = None
[docs] def restart(self, timeout):
'''
Restart the timer.
:param float/int timeout: A timer delay in seconds.
'''
if self.is_started():
self.stop()
self.start(timeout)
[docs] def is_started(self):
'''
T.is_started() -> boolean
Return `True` is the timer is started otherwise returns `False`.
'''
return self.Task is not None
async def _job(self, timeout):
await asyncio.sleep(timeout)
self.Task = None
if self.AutoRestart:
self.start(timeout)
await self.Handler()
def _on_stop(self, message_type, n):
# This is to ensure timer stop on application exit
self.stop()