rewrite of the repeat scheduler and implement cancel()

weboob/core/scheduler.py

@@ -18,7 +18,7 @@
 
 from __future__ import with_statement
 
-from threading import Timer, Event, RLock
+from threading import Timer, Event, RLock, _Timer
 from weboob.tools.log import getLogger
 
 
@@ -32,12 +32,22 @@ class IScheduler(object):
     def repeat(self, interval, function, *args):
         raise NotImplementedError()
 
+    def cancel(self, ev):
+        raise NotImplementedError()
+
     def run(self):
         raise NotImplementedError()
 
     def want_stop(self):
         raise NotImplementedError()
 
+class RepeatedTimer(_Timer):
+    def run(self):
+        while not self.finished.is_set():
+            self.function(*self.args, **self.kwargs)
+            self.finished.wait(self.interval)
+        self.finished.set()
+
 class Scheduler(IScheduler):
     def __init__(self):
         self.logger = getLogger('scheduler')
@@ -47,27 +57,43 @@ class Scheduler(IScheduler):
         self.queue = {}
 
     def schedule(self, interval, function, *args):
-        if self.stop_event.isSet():
+        return self._schedule(Timer, interval, self._schedule_callback, function, *args)
+
+    def repeat(self, interval, function, *args):
+        return self._schedule(RepeatedTimer, interval, self._repeat_callback, function, *args)
+
+    def _schedule(self, klass, interval, meta_func, function, *args):
+        if self.stop_event.is_set():
             return
 
         with self.mutex:
             self.count += 1
             self.logger.debug('function "%s" will be called in %s seconds' % (function.__name__, interval))
-            timer = Timer(interval, self._callback, (self.count, function, args))
+            timer = klass(interval, meta_func, (self.count, interval, function, args))
             self.queue[self.count] = timer
             timer.start()
             return self.count
 
-    def _callback(self, count, function, args):
+    def _schedule_callback(self, count, interval, function, args):
         with self.mutex:
             self.queue.pop(count)
         return function(*args)
 
-    def repeat(self, interval, function, *args):
+    def _repeat_callback(self, count, interval, function, args):
-        return self._repeat(True, interval, function, *args)
+        function(*args)
+        with self.mutex:
+            e = self.queue[count]
+            self.logger.debug('function "%s" will be called in %s seconds' % (function.__name__, e.interval))
 
-    def _repeat(self, first, interval, function, *args):
+    def cancel(self, ev):
-        return self.schedule(0 if first else interval, self._repeated_cb, interval, function, args)
+        with self.mutex:
+            try:
+                e = self.queue.pop(ev)
+            except KeyError:
+                return False
+            e.cancel()
+            self.logger.debug('scheduled function "%s" is canceled' % e.function)
+            return True
 
     def _wait_to_stop(self):
         self.want_stop()
@@ -96,7 +122,3 @@ class Scheduler(IScheduler):
                 # Contrary to _wait_to_stop(), don't call t.join
                 # because want_stop() have to be non-blocking.
             self.queue = {}
-
-    def _repeated_cb(self, interval, function, args):
-        function(*args)
-        self._repeat(False, interval, function, *args)