inefficient...

Author: r3w0p

bobocep/cep/engine/engine.py

@@ -6,7 +6,7 @@
 CEP engine.
 """
 
-from queue import Queue
+from queue import Queue, Full as QueueFull
 from threading import RLock
 from typing import Any, List, Optional, Tuple
 
@@ -25,6 +25,7 @@
 
 _EXC_QUEUE_FULL_IN = "incoming queue is full (max size: {})"
 _EXC_QUEUE_FULL_OUT = "outgoing queue is full (max size: {})"
+_EXC_RUNSERIAL_PHENOM_N0T_FOUND = "run serial phenomenon name {} not found"
 
 
 class BoboEngineError(BoboError):
@@ -46,8 +47,7 @@ def __init__(
             gen_data: Optional[BoboGenData] = None,
             max_size_in: int = 0,
             max_size_out: int = 0,
-            fail_on_queue_full: bool = False,
-            empty_queue_priority: bool = True,
+            fail_on_queue_full: bool = True,
             actions_on_local_only: bool = True
     ):
         super().__init__()
@@ -62,10 +62,11 @@ def __init__(
         self._gen_data: Optional[BoboGenData] = gen_data
 
         self._queue_in: Queue[Any] = Queue(max(0, max_size_in))
-        self._queue_out: Queue[Tuple[BoboRunSerial, bool]] = Queue(max(0, max_size_out))
+        self._queue_out: Queue[Tuple[BoboRunSerial, bool]] = (
+            Queue(max(0, max_size_out))
+        )
 
         self._fail_on_queue_full: bool = fail_on_queue_full
-        self._empty_queue_priority: bool = empty_queue_priority
         self._local_only: bool = actions_on_local_only
 
         self._decider.subscribe(self)
@@ -103,29 +104,8 @@ def run(self) -> None:
         """
         Runs the engine until it is closed. This is a blocking operation.
         """
-        while True:
-            with self._lock:
-                if self._closed:
-                    return
-
-                self.update()
-
-    def add_data(self, data: Any) -> None:
-        """
-        :param data: Data to add to the incoming queue.
-
-        :raises BoboEngineError: Incoming queue is full.
-        """
-        with self._lock:
-            if self._closed:
-                return
-
-            if not self._queue_in.full():
-                self._queue_in.put(data)
-
-            elif self._fail_on_queue_full:
-                raise BoboEngineError(
-                    _EXC_QUEUE_FULL_IN.format(self._queue_in.maxsize))
+        while self.update():
+            pass
 
     def update(self) -> bool:
         """
@@ -141,56 +121,35 @@ def update(self) -> bool:
             if self._closed:
                 return False
 
-        # Receiver
-        self._rec_update()
-        if self._empty_queue_priority:
-            while self._rec_update():
-                pass
-
-        # Producer
-        self._pro_update()
-        if self._empty_queue_priority:
-            while not self._queue_out.empty():
-                self._pro_update()
+            self._rec_dec_process()
+            self._pro_process()
+            self._for_process()
 
-        # Forwarder
-        self._for_update()
-        if self._empty_queue_priority:
-            while not self._handler.empty():
-                self._for_update()
-
-        return True
-
-    def _rec_update(self) -> bool:
-        data_in = None
-
-        # Process data from incoming queue
-        if not self._queue_in.empty():
-            data_in = self._queue_in.get_nowait()
-            self._rec_update_process(data_in)
-
-        # Process data from data generator
-        if self._gen_data is not None:
-            data_gen = self._gen_data.maybe_generate()
+            return True
 
-            if data_gen is not None:
-                self._rec_update_process(data_gen)
+    def add_data(self, data: Any) -> None:
+        """
+        :param data: Data to add to the incoming queue.
 
-        return data_in is not None
+        :raises BoboEngineError: Incoming queue is full.
+        """
+        with self._lock:
+            if self._closed:
+                return
 
-    def _rec_update_process(self, data: Any) -> bool:
         # Ensure data are valid
-        if self._validator is not None and not self._validator.is_valid(data):
-            for s in self._subscribers:
-                s.on_invalid_data_received(data)
-
-            return False
-
-        for s in self._subscribers:
-            s.on_valid_data_received(data)
-
-        # Pass to decider
-        return self.decider.process(data)
+        if (
+                self._validator is not None and
+                not self._validator.is_valid(data)
+        ):
+            return
+
+        try:
+            self._queue_in.put_nowait(data)
+        except QueueFull:
+            if self._fail_on_queue_full:
+                raise BoboEngineError(
+                    _EXC_QUEUE_FULL_IN.format(self._queue_in.maxsize))
 
     def on_decider_update(
             self,
@@ -202,14 +161,37 @@ def on_decider_update(
             if self._closed:
                 return
 
-        for run in completed:
-            if not self._queue_out.full():
-                self._queue_out.put((run, local))
-            else:
-                raise BoboEngineError(
-                    _EXC_QUEUE_FULL_OUT.format(self._queue_out.maxsize))
+        for runserial in completed:
+            try:
+                self._queue_out.put_nowait((runserial, local))
+            except QueueFull:
+                if self._fail_on_queue_full:
+                    raise BoboEngineError(
+                        _EXC_QUEUE_FULL_OUT.format(self._queue_out.maxsize))
+
+    def _rec_dec_process(self) -> bool:
+        data_in = None
+        data_gen = None
 
-    def _pro_update(self) -> bool:
+        # Process data from incoming queue
+        if not self._queue_in.empty():
+            data_in = self._queue_in.get_nowait()
+
+            for s in self._subscribers:
+                s.on_valid_data_received(data_in)
+
+            self.decider.process(data_in)
+
+        # Process data from data generator
+        if self._gen_data is not None:
+            data_gen = self._gen_data.maybe_generate()
+
+            if data_gen is not None:
+                self.decider.process(data_gen)
+
+        return data_in is not None or data_gen is not None
+
+    def _pro_process(self) -> bool:
         # Process data from outgoing queue
         if not self._queue_out.empty():
             runserial, local = self._queue_out.get_nowait()
@@ -218,7 +200,9 @@ def _pro_update(self) -> bool:
                 runserial.phenomenon_name)
 
             if phenom is None:
-                raise BoboEngineError(runserial.phenomenon_name)
+                raise BoboEngineError(
+                    _EXC_RUNSERIAL_PHENOM_N0T_FOUND.format(
+                        runserial.phenomenon_name))
 
             event = BoboEventComplex(
                 event_id=self._decider.gen_event_id.generate(),
@@ -241,7 +225,7 @@ def _pro_update(self) -> bool:
 
         return False
 
-    def _for_update(self) -> bool:
+    def _for_process(self) -> bool:
         hres: Optional[BoboHandlerResponse] = \
             self._handler.get_handler_response()
 

bobocep/cep/engine/pubsub.py

@@ -21,12 +21,6 @@ def on_valid_data_received(self, data: Any) -> None:
         :param data: Valid data received by engine.
         """
 
-    @abstractmethod
-    def on_invalid_data_received(self, data: Any) -> None:
-        """
-        :param data: Invalid data received by engine.
-        """
-
 
 class BoboEnginePublisher(ABC):
     """