| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566 |
- from __future__ import absolute_import
- import asyncio
- from functools import wraps, partial
- from apscheduler.schedulers.base import BaseScheduler
- from apscheduler.util import maybe_ref
- def run_in_event_loop(func):
- @wraps(func)
- def wrapper(self, *args, **kwargs):
- wrapped = partial(func, self, *args, **kwargs)
- self._eventloop.call_soon_threadsafe(wrapped)
- return wrapper
- class AsyncIOScheduler(BaseScheduler):
- """
- A scheduler that runs on an asyncio (:pep:`3156`) event loop.
- The default executor can run jobs based on native coroutines (``async def``).
- Extra options:
- ============== =============================================================
- ``event_loop`` AsyncIO event loop to use (defaults to the global event loop)
- ============== =============================================================
- """
- _eventloop = None
- _timeout = None
- def start(self, paused=False):
- if not self._eventloop:
- self._eventloop = asyncio.get_event_loop()
- super(AsyncIOScheduler, self).start(paused)
- @run_in_event_loop
- def shutdown(self, wait=True):
- super(AsyncIOScheduler, self).shutdown(wait)
- self._stop_timer()
- def _configure(self, config):
- self._eventloop = maybe_ref(config.pop('event_loop', None))
- super(AsyncIOScheduler, self)._configure(config)
- def _start_timer(self, wait_seconds):
- self._stop_timer()
- if wait_seconds is not None:
- self._timeout = self._eventloop.call_later(wait_seconds, self.wakeup)
- def _stop_timer(self):
- if self._timeout:
- self._timeout.cancel()
- del self._timeout
- @run_in_event_loop
- def wakeup(self):
- self._stop_timer()
- wait_seconds = self._process_jobs()
- self._start_timer(wait_seconds)
- def _create_default_executor(self):
- from apscheduler.executors.asyncio import AsyncIOExecutor
- return AsyncIOExecutor()
|
