pubsub

Author: r3w0p

bobocep/cep/engine/engine.py

@@ -181,8 +181,14 @@ def _rec_update(self) -> bool:
     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)
 

bobocep/cep/engine/pubsub.py

@@ -7,13 +7,26 @@
 """
 
 from abc import ABC, abstractmethod
+from typing import Any
 
 
 class BoboEngineSubscriber(ABC):
     """
     Engine subscriber interface.
     """
 
+    @abstractmethod
+    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):
     """