Remove schedualer

Author: Inbal Tako

securenative/event_manager.py

@@ -1,6 +1,5 @@
 import copy
 import json
-import sched
 import threading
 import time
 
@@ -27,17 +26,16 @@ def __init__(self, options=SecureNativeOptions(), http_client=None):
         else:
             self.http_client = http_client
 
+        self.thread = threading.Thread(target=self.run, daemon=True)
+        self.thread.start()
+
         self.options = options
         self.queue = list()
         self.send_enabled = False
         self.attempt = 0
         self.coefficients = [1, 1, 2, 3, 5, 8, 13]
-        self.scheduler = None
         self.thread = None
-
-        if self.options.auto_send and not self.options.disable:
-            interval_seconds = max(options.interval // 1000, 1)
-            threading.Timer(interval_seconds, self.flush).start()
+        self.interval = options.interval
 
     def send_async(self, event, resource_path):
         if self.options.disable:
@@ -50,7 +48,7 @@ def send_async(self, event, resource_path):
             False
         )
 
-        self.queue.insert(0, item)
+        self.queue.append(item)
         if self._is_queue_full():
             self.queue = self.queue[:len(self.queue - 1)]
 
@@ -79,51 +77,47 @@ def send_sync(self, event, resource_path, retry):
                 json.dumps(EventManager.serialize(event)),
                 retry
             )
-            self.queue.insert(0, item)
+            self.queue.append(0, item)
             if self._is_queue_full():
                 self.queue = self.queue[:len(self.queue - 1)]
         return res
 
     def _is_queue_full(self):
         return len(self.queue) > self.options.max_events
 
-    def _send_events(self):
-        if len(self.queue) > 0 and self.send_enabled:
-            for item in self.queue:
-                try:
-                    res = self.http_client.post(item.url, item.body)
-                    if res.status_code is 401:
-                        item.retry = False
-                    elif res.status_code != 200:
-                        raise SecureNativeHttpException(res.status_code)
-
-                    Logger.debug("Event successfully sent; {}".format(item.body))
-                    self.queue.remove(item)
-                    return res
-                except Exception as e:
-                    Logger.error("Failed to send event; {}".format(e))
-                    if item.retry:
-                        if len(self.coefficients) == self.attempt + 1:
-                            self.attempt = 0
-
-                        back_off = self.coefficients[self.attempt] * self.options.interval
-                        Logger.debug("Automatic back-off of {}".format(back_off))
-                        self.send_enabled = False
-                        time.sleep(back_off)
-                        self.send_enabled = True
-                    else:
+    def run(self):
+        while True:
+            if len(self.queue) > 0 and self.send_enabled:
+                for item in self.queue:
+                    try:
+                        res = self.http_client.post(item.url, item.body)
+                        if res.status_code is 401:
+                            item.retry = False
+                        elif res.status_code != 200:
+                            raise SecureNativeHttpException(res.status_code)
+
+                        Logger.debug("Event successfully sent; {}".format(item.body))
                         self.queue.remove(item)
+                        return res
+                    except Exception as e:
+                        Logger.error("Failed to send event; {}".format(e))
+                        if item.retry:
+                            if len(self.coefficients) == self.attempt + 1:
+                                self.attempt = 0
+
+                            back_off = self.coefficients[self.attempt] * self.options.interval
+                            Logger.debug("Automatic back-off of {}".format(back_off))
+                            self.send_enabled = False
+                            time.sleep(back_off)
+                            self.send_enabled = True
+                        else:
+                            self.queue.remove(item)
+                time.sleep(self.interval/1000)
 
     def start_event_persist(self):
         Logger.debug("Starting automatic event persistence")
         if self.options.auto_send or self.send_enabled:
             self.send_enabled = True
-            try:
-                self.scheduler = sched.scheduler(time.time, time.sleep)
-                self.scheduler.enter(self.options.interval, 1, self._send_events)
-                self.thread = threading.Thread(target=self.scheduler.run).start()
-            except Exception as e:
-                Logger.error("Could not start event scheduler; {}".format(e))
         else:
             Logger.debug("Automatic event persistence is disabled, you should persist events manually")
 
@@ -133,7 +127,6 @@ def stop_event_persist(self):
             try:
                 if self.thread:
                     self.thread.stop()
-                    self.scheduler.cancel(self._send_events)
             except ValueError as e:
                 Logger.error("Could not stop event scheduler; {}".format(e))