add timing test

timing.py

@@ -1,24 +1,32 @@
 #!/usr/bin/env python
 # -*- coding:utf-8 -*-
 import datetime
-from threading import Lock
-
+import threading
 import time
+import uuid
+import logging
+
+
+def generate_task_id():
+    return uuid.uuid1().hex
 
 
 class Timing:
-    def __init__(self, sleep_interval=3):
+    def __init__(self, sleep_interval=1):
         self.task = {}
-        self.lock = Lock()
+        self.lock = threading.Lock()
         self.sleep_interval = sleep_interval
 
-    def add_task(self, task_id, name, interval, func, *args, **kwargs):
+    def add_task(self, name, interval, func, count, task_id=generate_task_id(), *args, **kwargs):
         """
         Add a timing task and schedule will execute it.
         :param task_id: Unique task id.
         :param name: Task name.
         :param interval: The interval between the next task execution.
         :param func: The function of timing task execution.
+        :param count: The total number of times the task is executed. If it is 0, there is no limit.
+        :param args:
+        :param kwargs:
         :return:
         """
         if not isinstance(task_id, str):
@@ -27,11 +35,16 @@ class Timing:
             raise TypeError('interval must be int')
         if interval < 0:
             raise ValueError('interval must be bigger than 0')
-        if not isinstance(func, function):
+        if not callable(func):
             raise TypeError('func must be func')
+        if not isinstance(count, int):
+            raise TypeError('count must be int')
+        if count < 0:
+            count = 0
 
         self.lock.acquire()
-        self.task[task_id] = {'name': name, 'interval': interval, 'func': func, 'args': args, 'kwargs': kwargs}
+        self.task[task_id] = {'name': name, 'interval': interval, 'func': func, 'count': count, '__count': 0,
+                              'args': args, 'kwargs': kwargs}
         self.lock.release()
 
     def delete_task(self, task_id):
@@ -55,22 +68,36 @@ class Timing:
 
     def run(self):
         while True:
+            clear_keys = []
             self.lock.acquire()
-            for task_id, task_detail in self.task.keys():
-                now = int(datetime.datetime.now().timestamp() * 1000)
-                interval = task_detail['interval']
-                if interval - (now % interval) > 1:
-                    continue
-                task_detail['func']()
+            for task_id in self.task.keys():
+                task_detail = self.task[task_id]
+                try:
+                    now = int(datetime.datetime.now().timestamp() * 1000)
+                    interval = task_detail['interval']
+                    if interval - (now % interval) > 1:
+                        continue
+                    t = threading.Thread(target=task_detail['func'], name=task_detail['name'])
+                    t.start()
+
+                    task_detail['__count'] = task_detail['__count'] + 1
+                    if task_detail['__count'] >= task_detail['count']:
+                        clear_keys.append(task_id)
+                except Exception as e:
+                    logging.error('[TIMING] timing meet error, id: %s, name: %s, error: %s',
+                                  task_id, task_detail['name'], e)
             self.lock.release()
+
+            for key in clear_keys:
+                self.delete_task(key)
             self.sleep()
 
 
 defaultTiming = Timing()
 
 
-def add_task(task_id, name, interval, func, *args, **kwargs):
-    defaultTiming.add_task(task_id, name, interval, func, *args, **kwargs)
+def add_task(name, interval, func, count, task_id=generate_task_id(), *args, **kwargs):
+    defaultTiming.add_task(name, interval, func, count, task_id, *args, **kwargs)
 
 
 def delete_task(task_id):

timing_test.py

@@ -2,10 +2,29 @@
 # -*- coding:utf-8 -*-
 
 import unittest
+import timing
+import time
+import threading
+
 
 class TimingTest(unittest.TestCase):
-    pass
+    @staticmethod
+    def print_1():
+        print(1)
+
+    def test_timing_run(self):
+        print('test_timing_run start')
+        p1 = self.print_1
+        timing.add_task('task1', 3, p1, 3)
+        t = threading.Thread(target=timing.run)
+        t.setDaemon(True)
+        t.start()
+        t.join(10)
+        # print('start')
+        # time.sleep(10)
+        # print('end')
+        self.assertEqual(len(timing.defaultTiming.task), 0)
 
 
 if __name__ == '__main__':
-    pass
+    unittest.main()