from datetime import datetime, timedelta

from apscheduler.scheduler import Scheduler
from apscheduler.jobstores.shelve_store import ShelveJobStore


def alarm(time):
    print('Alarm! This alarm was scheduled at %s.' % time)


if __name__ == '__main__':
    scheduler = Scheduler(standalone=True)
    scheduler.add_jobstore(ShelveJobStore('example.db'), 'shelve')
    alarm_time = datetime.now() + timedelta(seconds=10)
    scheduler.add_date_job(alarm, alarm_time, name='alarm',
                           jobstore='shelve', args=[datetime.now()])
    print('To clear the alarms, delete the example.db file.')
    print('Press Ctrl+C to exit')
    try:
        scheduler.start()
    except (KeyboardInterrupt, SystemExit):
        pass

class Scheduler(object):
    """
    This class is responsible for scheduling jobs and triggering
    their execution.
    """

    _stopped = True
    _thread = None

    def __init__(self, gconfig={}, **options):
        self._wakeup = Event()
        self._jobstores = {}
        self._jobstores_lock = Lock()
        self._listeners = []
        self._listeners_lock = Lock()
        self._pending_jobs = []
        self.configure(gconfig, **options)

def add_jobstore(self, jobstore, alias, quiet=False):
    self._jobstores_lock.acquire() # 请求锁
    try:
        if alias in self._jobstores:
            raise KeyError('Alias "%s" is already in use' % alias)
        self._jobstores[alias] = jobstore # 将 jobstore 别名作为 key，添加到 self._jobstores 中
        jobstore.load_jobs() # 加载 jobstore 中所有 job
    finally:
        self._jobstores_lock.release() # 释放锁

    # Notify listeners that a new job store has been added
    self._notify_listeners(JobStoreEvent(EVENT_JOBSTORE_ADDED, alias)) # 事件通知

    # Notify the scheduler so it can scan the new job store for jobs
    if not quiet:
        self._wakeup.set() # 将 Event 置为 True

def add_date_job(self, func, date, args=None, kwargs=None, **options):
    trigger = SimpleTrigger(date)
    return self.add_job(trigger, func, args, kwargs, **options)

def add_job(self, trigger, func, args, kwargs, jobstore='default',
            **options):
    job = Job(trigger, func, args or [], kwargs or {},
              options.pop('misfire_grace_time', self.misfire_grace_time),
              options.pop('coalesce', self.coalesce), **options) # 将 job 实例化
    if not self.running: # 如果 scheduler 未启动，那么将其添加到等待队列中
        self._pending_jobs.append((job, jobstore))
        logger.info('Adding job tentatively -- it will be properly '
                    'scheduled when the scheduler starts')
    else:
        self._real_add_job(job, jobstore, True) # 否则添加 job 到 jobstore 中
    return job

def _real_add_job(self, job, jobstore, wakeup):
    job.compute_next_run_time(datetime.now()) # 计算job 下次运行时间
    if not job.next_run_time:
        raise ValueError('Not adding job since it would never be run')

    self._jobstores_lock.acquire() # 请求锁
    try:
        try:
            store = self._jobstores[jobstore]
        except KeyError:
            raise KeyError('No such job store: %s' % jobstore)
        store.add_job(job) # 添加 job
    finally:
        self._jobstores_lock.release()

    # Notify listeners that a new job has been added
    event = JobStoreEvent(EVENT_JOBSTORE_JOB_ADDED, jobstore, job)
    self._notify_listeners(event) # 事件通知

    logger.info('Added job "%s" to job store "%s"', job, jobstore)

    # Notify the scheduler about the new job
    if wakeup:
        self._wakeup.set() # # 将 Event 置为 True

def start(self):
    ...
    # Schedule all pending jobs
    for job, jobstore in self._pending_jobs: # 将 scheduler 未运行时添加的 job，即在等待队列中的 job 添加到 jobstore 中
        self._real_add_job(job, jobstore, False)
    del self._pending_jobs[:]

    self._stopped = False
    if self.standalone:
        self._main_loop()
    else:
        self._thread = Thread(target=self._main_loop, name='APScheduler')
        self._thread.setDaemon(self.daemonic)
        self._thread.start()

def _main_loop(self):
    """Executes jobs on schedule."""

    logger.info('Scheduler started')
    self._notify_listeners(SchedulerEvent(EVENT_SCHEDULER_START)) # 事件通知 

    self._wakeup.clear()
    while not self._stopped:
        logger.debug('Looking for jobs to run')
        now = datetime.now()
        next_wakeup_time = self._process_jobs(now) # 计算下次唤醒时间

        if next_wakeup_time is not None:
            wait_seconds = time_difference(next_wakeup_time, now)
            logger.debug('Next wakeup is due at %s (in %f seconds)',
                         next_wakeup_time, wait_seconds)
            try:
                self._wakeup.wait(wait_seconds) # 等待 Event flag
            except IOError:  # Catch errno 514 on some Linux kernels
                pass
            self._wakeup.clear()
        elif self.standalone:
            logger.debug('No jobs left; shutting down scheduler')
            self.shutdown() # 若 scheduler standalone 为 True 且 jobs 为空，则停止 scheduler
            break
        else:
            logger.debug('No jobs; waiting until a job is added')
            try:
                self._wakeup.wait() # 等待 Event flag
            except IOError:  # Catch errno 514 on some Linux kernels
                pass
            self._wakeup.clear()

class JobStore(object):
    def add_job(self, job):
    def update_job(self, job):
    def remove_job(self, job):
    def load_jobs(self):
    def close(self):

class JobEvent(SchedulerEvent):
    def __init__(self, code, job, scheduled_run_time, retval=None,
                 exception=None, traceback=None):
        SchedulerEvent.__init__(self, code)
        self.job = job
        self.scheduled_run_time = scheduled_run_time
        self.retval = retval
        self.exception = exception
        self.traceback = traceback

from huey import RedisHuey

huey = RedisHuey('simple.test', blocking=True)

@huey.task()
def count_beans(num):
    print('-- counted %s beans --' % num)
    return 'Counted %s beans' % num

class RedisHuey(Huey):
    def get_storage(self, read_timeout=1, max_errors=1000,
                    connection_pool=None, url=None, **connection_params):
        return RedisStorage(
            name=self.name,
            blocking=self.blocking,
            read_timeout=read_timeout,
            max_errors=max_errors,
            connection_pool=connection_pool,
            url=url,
            **connection_params)

class RedisStorage(BaseStorage):
    redis_client = Redis

    def __init__(self, name='huey', blocking=False, read_timeout=1,
                 max_errors=1000, connection_pool=None, url=None,
                 client_name=None, **connection_params):

        if Redis is None:
            raise ImportError('"redis" python module not found, cannot use '
                              'Redis storage backend. Run "pip install redis" '
                              'to install.')

        if sum(1 for p in (url, connection_pool, connection_params) if p) > 1:
            raise ValueError(
                'The connection configuration is over-determined. '
                'Please specify only one of the the following: '
                '"url", "connection_pool", or "connection_params"')

        if url:
            connection_pool = ConnectionPool.from_url(
                url, decode_components=True)
        elif connection_pool is None:
            connection_pool = ConnectionPool(**connection_params)
    ...

    def clean_name(self, name):
    def convert_ts(self, ts):
    def enqueue(self, data):
    def dequeue(self):
    ...
    def get_errors(self, limit=None, offset=0):
    def flush_errors(self):
    def emit(self, message):
    def listener(self):
    def __iter__(self):

def task(self, retries=0, retry_delay=0, retries_as_argument=False,
         include_task=False, name=None, **task_settings):
    def decorator(func):
        """
        Decorator to execute a function out-of-band via the consumer.
        """
        return TaskWrapper(
            self,
            func.func if isinstance(func, TaskWrapper) else func,
            retries=retries,
            retry_delay=retry_delay,
            retries_as_argument=retries_as_argument,
            include_task=include_task,
            name=name,
            **task_settings)
    return decorator

class TaskWrapper(object):
    def __init__(self, huey, func, retries=0, retry_delay=0,
                 retries_as_argument=False, include_task=False, name=None,
                 task_base=None, **task_settings):
        ...
        self.task_class = create_task(
            QueueTask if task_base is None else task_base,
            func,
            retries_as_argument,
            name,
            include_task,
            **task_settings)
        self.huey.registry.register(self.task_class)

    def is_revoked(self, dt=None, peek=True):
        return self.huey.is_revoked(self.task_class, dt, peek)

    def revoke(self, revoke_until=None, revoke_once=False):
        self.huey.revoke_all(self.task_class, revoke_until, revoke_once)

def consumer_main():
    ...
    huey_instance = load_huey(args[0])
    config.setup_logger()
    consumer = huey_instance.create_consumer(**config.values)
    consumer.run()
if __name__ == '__main__':
    consumer_main()

class Consumer(object):
    """
    Consumer sets up and coordinates the execution of the workers and scheduler
    and registers signal handlers.
    """

def loop(self, now=None):
    task = None
    exc_raised = True
    try:
        task = self.huey.dequeue() # 从 hue 中获取任务，也就是从 redis 中获取任务
        ...
    if task:
        self.delay = self.default_delay
        self.handle_task(task, now or self.get_now()) # 如果获取到了任务，对任务进行处理
    elif exc_raised or not self.huey.blocking:
        self.sleep()

def handle_task(self, task, ts):
    if not self.huey.ready_to_run(task, ts):
        self.add_schedule(task) # 若任务没有到达执行时间，则添加到 schedule 中
    elif not self.is_revoked(task, ts):
        self.process_task(task, ts) # 若任务没有被取消，则执行任务
    else:
        self.huey.emit_task(
            EVENT_REVOKED,
            task,
            timestamp=ts)
        self._logger.debug('Task %s was revoked, not running', task) # 任务被取消，通知 event

def process_task(self, task, ts):
    ...
    self.run_pre_execute_hooks(task)   # 执行 pre hook 动作
    ...  
    task_value = self.huey.execute(task) # 具体执行 task
    ...
    self.run_post_execute_hooks(task, task_value, exception) # 执行 post hook 动作
    ...

def execute(self, task):
    if not isinstance(task, QueueTask):
        raise TypeError('Unknown object: %s' % task)

    try:
        result = task.execute()

def create_task(task_class, func, retries_as_argument=False, task_name=None,
                include_task=False, **kwargs):
    def execute(self):
        args, kwargs = self.data or ((), {})
        if retries_as_argument:
            kwargs['retries'] = self.retries
        if include_task:
            kwargs['task'] = self
        return func(*args, **kwargs)

    attrs = {
        'execute': execute,
        '__module__': func.__module__,
        '__doc__': func.__doc__}
    attrs.update(kwargs)

    if not task_name:
        task_name = func.__name__

    return type(task_name, (task_class,), attrs)

class TaskLock(object):
    """
    Utilize the Storage key/value APIs to implement simple locking. For more
    details see :py:meth:`Huey.lock_task`.
    """
    def __init__(self, huey, name):
        self._huey = huey
        self._name = name
        self._key = '%s.lock.%s' % (self._huey.name, self._name)
        self._huey._locks.add(self._key)

    def __call__(self, fn):
        @wraps(fn)
        def inner(*args, **kwargs):
            with self:
                return fn(*args, **kwargs)
        return inner

    def __enter__(self):
        if not self._huey._put_if_empty(self._key, '1'):
            raise TaskLockedException('unable to set lock: %s' % self._name)

    def __exit__(self, exc_type, exc_val, exc_tb):
        self._huey._get_data(self._key)

def put_if_empty(self, key, value):
    return self.conn.hsetnx(self.result_key, key, value)

root@yiran-30-250:/usr/lib/systemd/system
 $ cat /root/project/watchdog/test.py
#!/usr/bin/python
# coding:utf-8
import os
import time
import socket
import logging

print("Test starting up...")
time.sleep(1)  # 模拟执行真实业务
print("Test startup finished")

try:
    sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
    addr = os.getenv("NOTIFY_SOCKET")  # systemd default addr = "/run/systemd/notify"
    if addr and addr[0] == "@":
        addr = "\0" + addr[1:]
except Exception:
    logging.error("Failed to get notify socket addr")

if sock and addr:
    sock.connect(addr)
    sock.sendall("READY=1")  # 通知 systemd 服务启动完成

count = 0

while True:
    print("Running...")
    time.sleep(2)
    if count <= 10:
        sock.sendall("WATCHDOG=1")  # 通知 systemd 服务正常运行
        logging.info("Notify socket addr is:%s", addr)
        logging.info("test.service watchdog timestamp update succeeded")
        count += 1

def countdown(n):
   while n > 0:
       yield n
       n -= 1
>>> for i in countdown(5):
...     print(i, end=' ')
...
5 4 3 2 1

In [7]: def recv_count():
   ...:     try:
   ...:         while True:
   ...:             n = yield
   ...:             print("T-minus", n)
   ...:     except GeneratorExit:
   ...:         print("boom")
   ...:
In [8]: r = recv_count()
In [9]: r.send(None) # init
In [10]: r.next() # yield 未接收到调用者发送具体值
('T-minus', None)
In [13]: r.send(1) 
('T-minus', 1)
In [14]: r.send(2)
('T-minus', 2)
In [15]: r.send(3)
('T-minus', 3)

root@yiran-30-250:/tmp
 $ cat a.py
#!/usr/bin/python3

def consumer():
    r = ''
    while True:
        n = yield r
        print('Consuming %s ...' % n)
        r = '200 OK'
def producer(c):
    c.send(None)
    n = 0
    while n < 5:
        n += 1
        print("Producing %s ..." % n)
        r = c.send(n)
        print("Consumer return: %s" % r)
    c.close()
c = consumer()
producer(c)

root@yiran-30-250:/tmp
 $ python3 a.py
Producing 1 ...
Consuming 1 ...
Consumer return: 200 OK
Producing 2 ...
Consuming 2 ...
Consumer return: 200 OK
Producing 3 ...
Consuming 3 ...
Consumer return: 200 OK
Producing 4 ...
Consuming 4 ...
Consumer return: 200 OK
Producing 5 ...
Consuming 5 ...
Consumer return: 200 OK