NovaBryan/EVE_Dump
Log | Files | 📦 下载最新源码 (.zip) | Refs | README

periodic_timer.py (873B)

      1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\datadog\threadstats\periodic_timer.py
      2 from threading import Thread, Event
      3 
      4 class PeriodicTimer(Thread):
      5 
      6     def __init__(self, interval, function, *args, **kwargs):
      7         Thread.__init__(self)
      8         self.daemon = True
      9         self.interval = interval
     10         self.function = function
     11         self.args = args
     12         self.kwargs = kwargs
     13         self.finished = Event()
     14 
     15     def _is_alive(self):
     16         return bool(self.finished) and bool(self.interval) and bool(self.function)
     17 
     18     def end(self):
     19         if self._is_alive():
     20             self.finished.set()
     21 
     22     def run(self):
     23         while True:
     24             if not self._is_alive() or self.finished.isSet():
     25                 break
     26             self.finished.wait(self.interval)
     27             self.function(*self.args, **self.kwargs)