Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _on_timer(self):
if loop.time() - self._last_time > self._timeout:
self.emit()
self.set_done()
else:
self._schedule()
def _schedule(self):
self._handle = loop.call_at(
self._last_time + self._timeout, self._on_timer)
ev.Timer(0.25, count=10).run()
->
[0.25, 0.5, 0.75, 1.0, 1.25, 1.5, 1.75, 2.0, 2.25, 2.5]
.. note::
When running inside a Jupyter notebook this will give an error
that the asyncio event loop is already running. This can be
remedied by applying
`nest_asyncio `_
or by using the top-level ``await`` statement of Jupyter::
await event.list()
"""
return loop.run_until_complete(self.list())
def emit_threadsafe(self, *args):
"""
Threadsafe version of :meth:`emit` that doesn't invoke the
listeners directly but via the event loop.
"""
loop.call_soon_threadsafe(self.emit, *args)