everything moved over but untested

- needs big clean-up
- improve interfaces
- lots of testing

new features going forward
- change phenomena dynamically (for non distributed only)

Author: r3w0p

bobocep/cep/action/handler.py

@@ -117,6 +117,10 @@ def get_handler_response(self) -> Optional[BoboHandlerResponse]:
                 return queue.get_nowait()
             return None
 
+    def empty(self) -> bool:
+        with self._lock:
+            return self._get_queue().empty()
+
     def size(self) -> int:
         """
         :return: The size of the handler queue.

bobocep/cep/engine/decider/decider.py

@@ -37,9 +37,7 @@ class BoboDeciderError(BoboEngineTaskError):
     """
 
 
-class BoboDecider(BoboDeciderPublisher,
-                  BoboReceiverSubscriber,
-                  BoboDistributedSubscriber):
+class BoboDecider(BoboDeciderPublisher, BoboDistributedSubscriber):
     """
     A decider task.
     """
@@ -85,6 +83,111 @@ def __init__(self,
         self._cache_halted: Optional[Deque[BoboRunSerial]] = \
             deque(maxlen=max_cache) if self._caching else None
 
+    @property
+    def gen_event_id(self):
+        """
+        :return: Event ID generator.
+        """
+        return self._gen_event_id
+
+    @property
+    def gen_run_id(self):
+        """
+        :return: Run ID generator.
+        """
+        return self._gen_run_id
+
+    @property
+    def gen_timestamp(self):
+        """
+        :return: Timestamp generator.
+        """
+        return self._gen_timestamp
+
+    @property
+    def phenomena(self) -> Tuple[BoboPhenomenon, ...]:
+        """
+        :return: All phenomena under consideration by the decider.
+        """
+        with self._lock:
+            return tuple(*self._phenomena.values())
+
+    def phenomenon(self, name: str) -> Optional[BoboPhenomenon]:
+        """
+        :param name: Name of the phenomenon.
+
+        :return: `BoboPhenomenon` that matches the name; `None` if not found.
+        """
+        with self._lock:
+            if name not in self._phenomena:
+                return None
+            return self._phenomena[name]
+
+    def all_runs(self) -> Tuple[BoboRun, ...]:
+        """
+        :return: All active runs in the decider.
+        """
+        with self._lock:
+            runs: List[BoboRun] = []
+            for phenomenon_name, dict_patterns in self._runs.items():
+                for pattern_name, dict_runs in dict_patterns.items():
+                    for _, drun in dict_runs.items():
+                        runs.append(drun)
+            return tuple(runs)
+
+    def runs_from(self,
+                  phenomenon_name: str,
+                  pattern_name: str) -> Tuple[BoboRun, ...]:
+        """
+        :param phenomenon_name: A phenomenon name.
+        :param pattern_name: A pattern name.
+        :return: The runs associated with the given
+            phenomenon and pattern name.
+        """
+        with self._lock:
+            if (
+                    phenomenon_name in self._runs and
+                    pattern_name in self._runs[phenomenon_name]
+            ):
+                return tuple(
+                    *self._runs[phenomenon_name][pattern_name].values()
+                )
+            return tuple()
+
+    def run_at(self,
+               phenomenon_name: str,
+               pattern_name: str,
+               run_id: str) -> Optional[BoboRun]:
+        """
+        :param phenomenon_name: A phenomenon name.
+        :param pattern_name: A pattern name.
+        :param run_id: A run ID.
+        :return: A run associated with the given phenomenon and pattern name;
+            or None if no such run exists.
+        """
+        with self._lock:
+            if (
+                    phenomenon_name in self._runs and
+                    pattern_name in self._runs[phenomenon_name] and
+                    run_id in self._runs[phenomenon_name][pattern_name]
+            ):
+                return self._runs[phenomenon_name][pattern_name][run_id]
+            return None
+
+    def close(self) -> None:
+        """
+        Closes the Decider.
+        """
+        with self._lock:
+            self._closed = True
+
+    def is_closed(self) -> bool:
+        """
+        :return: `True` if decider is set to close; `False` otherwise.
+        """
+        with self._lock:
+            return self._closed
+
     def subscribe(self, subscriber: BoboDeciderSubscriber) -> None:
         """
         :param subscriber: Subscriber to Decider data.
@@ -406,78 +509,6 @@ def _get_pattern(self,
                     return pattern
         return None
 
-    def phenomena(self) -> Tuple[BoboPhenomenon, ...]:
-        """
-        :return: All phenomena under consideration by the decider.
-        """
-        with self._lock:
-            return tuple(*self._phenomena.values())
-
-    def all_runs(self) -> Tuple[BoboRun, ...]:
-        """
-        :return: All active runs in the decider.
-        """
-        with self._lock:
-            runs: List[BoboRun] = []
-            for phenomenon_name, dict_patterns in self._runs.items():
-                for pattern_name, dict_runs in dict_patterns.items():
-                    for _, drun in dict_runs.items():
-                        runs.append(drun)
-            return tuple(runs)
-
-    def runs_from(self,
-                  phenomenon_name: str,
-                  pattern_name: str) -> Tuple[BoboRun, ...]:
-        """
-        :param phenomenon_name: A phenomenon name.
-        :param pattern_name: A pattern name.
-        :return: The runs associated with the given
-            phenomenon and pattern name.
-        """
-        with self._lock:
-            if (
-                    phenomenon_name in self._runs and
-                    pattern_name in self._runs[phenomenon_name]
-            ):
-                return tuple(
-                    *self._runs[phenomenon_name][pattern_name].values()
-                )
-            return tuple()
-
-    def run_at(self,
-               phenomenon_name: str,
-               pattern_name: str,
-               run_id: str) -> Optional[BoboRun]:
-        """
-        :param phenomenon_name: A phenomenon name.
-        :param pattern_name: A pattern name.
-        :param run_id: A run ID.
-        :return: A run associated with the given phenomenon and pattern name;
-            or None if no such run exists.
-        """
-        with self._lock:
-            if (
-                    phenomenon_name in self._runs and
-                    pattern_name in self._runs[phenomenon_name] and
-                    run_id in self._runs[phenomenon_name][pattern_name]
-            ):
-                return self._runs[phenomenon_name][pattern_name][run_id]
-            return None
-
-    def close(self) -> None:
-        """
-        Closes the Decider.
-        """
-        with self._lock:
-            self._closed = True
-
-    def is_closed(self) -> bool:
-        """
-        :return: `True` if decider is set to close; `False` otherwise.
-        """
-        with self._lock:
-            return self._closed
-
     def _process_event(self, data: Any, timestamp: int) -> \
             Tuple[List[BoboRun], List[BoboRun], List[BoboRun]]:
         """