diff options
Diffstat (limited to 'sleekxmpp/xmlstream/scheduler.py')
-rw-r--r-- | sleekxmpp/xmlstream/scheduler.py | 68 |
1 files changed, 36 insertions, 32 deletions
diff --git a/sleekxmpp/xmlstream/scheduler.py b/sleekxmpp/xmlstream/scheduler.py index 0e3d30b3..0af1f508 100644 --- a/sleekxmpp/xmlstream/scheduler.py +++ b/sleekxmpp/xmlstream/scheduler.py @@ -104,7 +104,7 @@ class Scheduler(object): quit -- Stop the scheduler. """ - def __init__(self, parentqueue=None): + def __init__(self, parentqueue=None, parentstop=None): """ Create a new scheduler. @@ -116,6 +116,7 @@ class Scheduler(object): self.thread = None self.run = False self.parentqueue = parentqueue + self.parentstop = parentstop def process(self, threaded=True): """ @@ -135,38 +136,41 @@ class Scheduler(object): def _process(self): """Process scheduled tasks.""" self.run = True - while self.run: - try: - wait = 1 - updated = False - if self.schedule: - wait = self.schedule[0].next - time.time() - try: - if wait <= 0.0: - newtask = self.addq.get(False) - else: - newtask = self.addq.get(True, wait) - except queue.Empty: - cleanup = [] - for task in self.schedule: - if time.time() >= task.next: - updated = True - if not task.run(): - cleanup.append(task) + try: + while self.run: + wait = 1 + updated = False + if self.schedule: + wait = self.schedule[0].next - time.time() + try: + if wait <= 0.0: + newtask = self.addq.get(False) else: - break - for task in cleanup: - x = self.schedule.pop(self.schedule.index(task)) - else: - updated = True - self.schedule.append(newtask) - finally: - if updated: - self.schedule = sorted(self.schedule, - key=lambda task: task.next) - except KeyboardInterrupt: - self.run = False - + newtask = self.addq.get(True, wait) + except queue.Empty: + cleanup = [] + for task in self.schedule: + if time.time() >= task.next: + updated = True + if not task.run(): + cleanup.append(task) + else: + break + for task in cleanup: + x = self.schedule.pop(self.schedule.index(task)) + else: + updated = True + self.schedule.append(newtask) + finally: + if updated: + self.schedule = sorted(self.schedule, + key=lambda task: task.next) + except KeyboardInterrupt: + self.run = False + if self.parentstop is not None: self.parentstop.set() + except SystemExit: + self.run = False + if self.parentstop is not None: self.parentstop.set() logging.debug("Qutting Scheduler thread") if self.parentqueue is not None: self.parentqueue.put(('quit', None, None)) |