refactor begin

- moved receiver into engine
- decides takes data directly and only produces simple events when needed
- problem with phenom.values() fixed
- removed kofi references
- updated requirements.txt

todo next
- move producer and forwarder into engine by reacting to decider's published data

Author: r3w0p

.github/FUNDING.yml

@@ -24,8 +24,6 @@ https://app.fossa.com/projects/git%2Bgithub.com%2Fr3w0p%2Fbobocep?ref=badge_shie
 
 [![JOSS](https://img.shields.io/badge/10.21105/joss.05858-red?label=JOSS)](
 https://doi.org/10.21105/joss.05858)
-[![Donate](https://img.shields.io/badge/ko--fi-red?label=donate)](
-https://ko-fi.com/r3w0p)
 <br/>
 
 `BoboCEP` is a [Complex Event Processing](https://en.wikipedia.org/wiki/Complex_event_processing) (CEP) engine

README.md

@@ -10,15 +10,16 @@
 from collections import deque
 from queue import Queue
 from threading import RLock
-from typing import Tuple, Dict, List, Optional, Deque
+from typing import Tuple, Dict, List, Optional, Deque, Any
 
 from bobocep.cep.engine.decider.pubsub import BoboDeciderPublisher, \
     BoboDeciderSubscriber
 from bobocep.cep.engine.decider.run import BoboRun
 from bobocep.cep.engine.decider.runserial import BoboRunSerial
 from bobocep.cep.engine.receiver.pubsub import BoboReceiverSubscriber
 from bobocep.cep.engine.task import BoboEngineTaskError, BoboEngineTask
-from bobocep.cep.event import BoboHistory, BoboEvent
+from bobocep.cep.event import BoboHistory, BoboEvent, BoboEventSimple
+from bobocep.cep.gen import BoboGenTimestamp
 from bobocep.cep.gen.event_id import BoboGenEventID
 from bobocep.cep.phenom.pattern.pattern import BoboPattern
 from bobocep.cep.phenom.phenom import BoboPhenomenon
@@ -36,8 +37,7 @@ class BoboDeciderError(BoboEngineTaskError):
     """
 
 
-class BoboDecider(BoboEngineTask,
-                  BoboDeciderPublisher,
+class BoboDecider(BoboDeciderPublisher,
                   BoboReceiverSubscriber,
                   BoboDistributedSubscriber):
     """
@@ -48,16 +48,14 @@ def __init__(self,
                  phenomena: List[BoboPhenomenon],
                  gen_event_id: BoboGenEventID,
                  gen_run_id: BoboGenEventID,
-                 max_cache: int = 0,
-                 max_size: int = 0):
+                 gen_timestamp: BoboGenTimestamp,
+                 max_cache: int = 0):
         """
         :param phenomena: List of phenomena.
         :param gen_event_id: Event ID generator.
         :param gen_run_id: Run ID generator.
         :param max_cache: Max cache size (<=0 means no caching).
             Default: 0.
-        :param max_size: Max queue size.
-            Default: 0 (unbounded).
         """
         super().__init__()
 
@@ -76,11 +74,10 @@ def __init__(self,
 
         self._gen_event_id: BoboGenEventID = gen_event_id
         self._gen_run_id: BoboGenEventID = gen_run_id
+        self._gen_timestamp: BoboGenTimestamp = gen_timestamp
         # Phenomenon Name => Pattern Name => Run ID => Run
         self._runs: Dict[str, Dict[str, Dict[str, BoboRun]]] = {}
         self._stub_history: BoboHistory = BoboHistory({})
-        self._max_size: int = max(0, max_size)
-        self._queue: Queue[BoboEvent] = Queue(self._max_size)
 
         self._caching: bool = max_cache > 0
         self._cache_completed: Optional[Deque[BoboRunSerial]] = \
@@ -96,7 +93,7 @@ def subscribe(self, subscriber: BoboDeciderSubscriber) -> None:
             if subscriber not in self._subscribers:
                 self._subscribers.append(subscriber)
 
-    def update(self) -> bool:
+    def process(self, data: Any) -> bool:
         """
         Performs an update cycle of the decider that takes an event from its
         queue and checks it against phenomena and existing runs.
@@ -108,10 +105,12 @@ def update(self) -> bool:
             if self._closed:
                 return False
 
-            if not self._queue.empty():
+            if data is not None:
+                timestamp: int = self._gen_timestamp.generate()
+
                 # Process event and collect changes to decider
                 rl_completed, rl_halted, rl_updated = \
-                    self._process_event(self._queue.get_nowait())
+                    self._process_event(data, timestamp)
 
                 completed: List[BoboRunSerial] = \
                     [run_c.serialize() for run_c in rl_completed]
@@ -200,20 +199,6 @@ def _maybe_cache(
             for h in halted:
                 self._cache_halted.append(h)
 
-    def on_receiver_update(self, event: BoboEvent) -> None:
-        """
-        :param event: Event from Receiver.
-        """
-        with self._lock:
-            if self._closed:
-                return
-
-            if not self._queue.full():
-                self._queue.put(event)
-            else:
-                raise BoboDeciderError(
-                    _EXC_QUEUE_FULL.format(self._max_size))
-
     def _maybe_check_against_cache(
             self,
             completed: List[BoboRunSerial],
@@ -380,7 +365,9 @@ def on_distributed_update(
                         phenomenon_name=runremote.phenomenon_name,
                         pattern=pattern,
                         block_index=runremote.block_index,
-                        history=runremote.history)
+                        history=runremote.history,
+                        gen_event_id=self._gen_event_id
+                    )
 
                     self._add_run(
                         runremote.phenomenon_name,
@@ -424,7 +411,7 @@ def phenomena(self) -> Tuple[BoboPhenomenon, ...]:
         :return: All phenomena under consideration by the decider.
         """
         with self._lock:
-            return tuple(self._phenomena.values())
+            return tuple(*self._phenomena.values())
 
     def all_runs(self) -> Tuple[BoboRun, ...]:
         """
@@ -453,7 +440,8 @@ def runs_from(self,
                     pattern_name in self._runs[phenomenon_name]
             ):
                 return tuple(
-                    self._runs[phenomenon_name][pattern_name].values())
+                    *self._runs[phenomenon_name][pattern_name].values()
+                )
             return tuple()
 
     def run_at(self,
@@ -476,13 +464,6 @@ def run_at(self,
                 return self._runs[phenomenon_name][pattern_name][run_id]
             return None
 
-    def size(self) -> int:
-        """
-        :return: The total number of events in the decider's queue.
-        """
-        with self._lock:
-            return self._queue.qsize()
-
     def close(self) -> None:
         """
         Closes the Decider.
@@ -497,19 +478,19 @@ def is_closed(self) -> bool:
         with self._lock:
             return self._closed
 
-    def _process_event(self, event: BoboEvent) -> \
+    def _process_event(self, data: Any, timestamp: int) -> \
             Tuple[List[BoboRun], List[BoboRun], List[BoboRun]]:
         """
         :param event: An event.
 
         :return: Runs that had a state change due to the event.
         """
-        r_halt_com, r_halt_incom, r_upd = self._check_against_runs(event)
-        p_halt_com, p_upd = self._check_against_patterns(event)
+        r_halt_com, r_halt_incom, r_upd = self._check_against_runs(data, timestamp)
+        p_halt_com, p_upd = self._check_against_patterns(data, timestamp)
 
         return (r_halt_com + p_halt_com), r_halt_incom, (r_upd + p_upd)
 
-    def _check_against_runs(self, event: BoboEvent) -> \
+    def _check_against_runs(self, data: Any, timestamp: int) -> \
             Tuple[List[BoboRun], List[BoboRun], List[BoboRun]]:
         """
         :param event: An event.
@@ -528,7 +509,7 @@ def _check_against_runs(self, event: BoboEvent) -> \
                     # If an internal state change occurs in the run...
                     run_eval: bool
                     try:
-                        run_eval = run.process(event)
+                        run_eval = run.process(data, timestamp)
                     except (Exception,):
                         continue
 
@@ -549,7 +530,7 @@ def _check_against_runs(self, event: BoboEvent) -> \
 
         return runs_halted_complete, runs_halted_incomplete, runs_updated
 
-    def _check_against_patterns(self, event: BoboEvent) -> \
+    def _check_against_patterns(self, data: Any, timestamp: int) -> \
             Tuple[List[BoboRun], List[BoboRun]]:
         """
         :param event: An event.
@@ -565,22 +546,34 @@ def _check_against_patterns(self, event: BoboEvent) -> \
                 any_eval: bool = False
                 for predicate in pattern.blocks[0].predicates:
                     try:
-                        if predicate.evaluate(event, self._stub_history):
+                        if predicate.evaluate(
+                                data, timestamp, self._stub_history
+                        ):
                             any_eval = True
                             break
                     except (Exception,):
                         pass
 
                 # ...create a run.
                 if any_eval:
+                    if isinstance(data, BoboEvent):
+                        event = data
+                    else:
+                        event = BoboEventSimple(
+                            event_id=self._gen_event_id.generate(),
+                            timestamp=timestamp,
+                            data=data)
+
                     newrun = BoboRun(
                         run_id=self._gen_run_id.generate(),
                         phenomenon_name=phenomenon.name,
                         pattern=pattern,
                         block_index=1,
                         history=BoboHistory({
                             pattern.blocks[0].group: [event]
-                        }))
+                        }),
+                        gen_event_id=self._gen_event_id
+                    )
 
                     if newrun.is_halted() and newrun.is_complete():
                         runs_halted_complete.append(newrun)