Event engine - Effect dispatcher (#160)

* Event Dispatcher

Author: Sebastian Molenda

pubnub/event_engine/__init__.py

@@ -0,0 +1,21 @@
+from pubnub.event_engine import effects
+
+
+class Dispatcher:
+    def __init__(self) -> None:
+        self._managed_effects = {}
+        self._effect_emitter = effects.EmitEffect()
+
+    def dispatch_effect(self, effect: effects.PNEffect):
+        if isinstance(effect, effects.PNEmittableEffect):
+            self._effect_emitter.emit(effect)
+
+        if isinstance(effect, effects.PNManageableEffect):
+            managed_effect = effects.ManagedEffect(effect)
+            managed_effect.run()
+            self._managed_effects[effect.__class__.__name__] = managed_effect
+
+        if isinstance(effect, effects.PNCancelEffect):
+            if effect.cancel_effect in self._managed_effects:
+                self._managed_effects[effect.cancel_effect].stop()
+                del self._managed_effects[effect.cancel_effect]

pubnub/event_engine/dispatcher.py

@@ -1,29 +1,38 @@
-from typing import Union
+from typing import List, Union
 from pubnub.exceptions import PubNubException
 from pubnub.enums import PNStatusCategory
+from pubnub.pubnub import PubNub
 
 
 class PNEffect:
     pass
 
 
-class HandshakeEffect(PNEffect):
-    def __init__(self, channels: Union[None, list[str]], groups: Union[None, list[str]]) -> None:
+class PNManageableEffect(PNEffect):
+    pass
+
+
+class PNCancelEffect(PNEffect):
+    cancel_effect: str
+
+
+class HandshakeEffect(PNManageableEffect):
+    def __init__(self, channels: Union[None, List[str]] = None, groups: Union[None, List[str]] = None) -> None:
         super().__init__()
         self.channels = channels
         self.groups = groups
 
 
-class CancelHandshakeEffect(PNEffect):
-    pass
+class CancelHandshakeEffect(PNCancelEffect):
+    cancel_effect = HandshakeEffect.__name__
 
 
-class ReceiveMessagesEffect(PNEffect):
+class ReceiveMessagesEffect(PNManageableEffect):
     def __init__(self,
-                 channels: Union[None, list[str]],
-                 groups: Union[None, list[str]],
-                 timetoken: Union[None, str],
-                 region: Union[None, int]
+                 channels: Union[None, List[str]] = None,
+                 groups: Union[None, List[str]] = None,
+                 timetoken: Union[None, str] = None,
+                 region: Union[None, int] = None
                  ) -> None:
         super().__init__()
         self.channels = channels
@@ -32,51 +41,95 @@ def __init__(self,
         self.region = region
 
 
-class CancelReceiveMessagesEffect(PNEffect):
-    pass
-
-
-class EmitMessagesEffect(PNEffect):
-    def __init__(self, messages: Union[None, list[str]]) -> None:
-        super().__init__()
-        self.messages = messages
-
-
-class EmitStatusEffect(PNEffect):
-    def __init__(self, status: Union[None, PNStatusCategory]) -> None:
-        super().__init__()
-        self.status = status
+class CancelReceiveMessagesEffect(PNCancelEffect):
+    cancel_effect = ReceiveMessagesEffect.__name__
 
 
-class HandshakeReconnectEffect(PNEffect):
+class HandshakeReconnectEffect(PNManageableEffect):
     def __init__(self,
-                 channels: Union[None, list[str]],
-                 groups: Union[None, list[str]],
-                 attempts: Union[None, int],
-                 reason: Union[None, PubNubException]
+                 channels: Union[None, List[str]] = None,
+                 groups: Union[None, List[str]] = None,
+                 attempts: Union[None, int] = None,
+                 reason: Union[None, PubNubException] = None
                  ) -> None:
         self.channels = channels
         self.groups = groups
         self.attempts = attempts
         self.reason = reason
 
 
-class CancelHandshakeEffect(PNEffect):
-    pass
+class CancelHandshakeReconnectEffect(PNCancelEffect):
+    cancel_effect = HandshakeReconnectEffect.__name__
 
 
-class ReceiveReconnectEffect(PNEffect):
+class ReceiveReconnectEffect(PNManageableEffect):
     def __init__(self,
-                 channels: Union[None, list[str]],
-                 groups: Union[None, list[str]],
-                 timetoken: Union[None, str],
-                 region: Union[None, int],
-                 attempts: Union[None, int],
-                 reason: Union[None, PubNubException]
+                 channels: Union[None, List[str]] = None,
+                 groups: Union[None, List[str]] = None,
+                 timetoken: Union[None, str] = None,
+                 region: Union[None, int] = None,
+                 attempts: Union[None, int] = None,
+                 reason: Union[None, PubNubException] = None
                  ) -> None:
         self.channels = channels
         self.groups = groups
         self.timetoken = timetoken
         self.region = region
         self.attempts = attempts
         self.reason = reason
+
+
+class CancelReceiveReconnectEffect(PNCancelEffect):
+    cancel_effect = ReceiveReconnectEffect.__name__
+
+
+class PNEmittableEffect(PNEffect):
+    pass
+
+
+class EmitMessagesEffect(PNEmittableEffect):
+    def __init__(self, messages: Union[None, List[str]]) -> None:
+        super().__init__()
+        self.messages = messages
+
+
+class EmitStatusEffect(PNEmittableEffect):
+    def __init__(self, status: Union[None, PNStatusCategory]) -> None:
+        super().__init__()
+        self.status = status
+
+
+class ManagedEffect:
+    pubnub: PubNub
+    effect: Union[PNManageableEffect, PNCancelEffect]
+
+    def set_pn(pubnub: PubNub):
+        pubnub = pubnub
+
+    def __init__(self, effect: Union[PNManageableEffect, PNCancelEffect]) -> None:
+        self.effect = effect
+
+    def run(self):
+        pass
+
+    def stop(self):
+        pass
+
+
+class EmitEffect:
+    pubnub: PubNub
+
+    def set_pn(pubnub: PubNub):
+        pubnub = pubnub
+
+    def emit(self, effect: PNEmittableEffect):
+        if isinstance(effect, EmitMessagesEffect):
+            self.emit_message(effect)
+        if isinstance(effect, EmitStatusEffect):
+            self.emit_status(effect)
+
+    def emit_message(self, effect: EmitMessagesEffect):
+        pass
+
+    def emit_status(self, effect: EmitStatusEffect):
+        pass

pubnub/event_engine/effects.py

@@ -1,4 +1,5 @@
 from pubnub.exceptions import PubNubException
+from typing import List
 
 
 class PNEvent:
@@ -18,18 +19,18 @@ def __init__(self, timetoken: str, region: int) -> None:
 
 
 class PNChannelGroupsEvent(PNEvent):
-    def __init__(self, channels: list[str], groups: list[str]) -> None:
+    def __init__(self, channels: List[str], groups: List[str]) -> None:
         self.channels = channels
         self.groups = groups
 
 
 class SubscriptionChangedEvent(PNChannelGroupsEvent):
-    def __init__(self, channels: list[str], groups: list[str]) -> None:
+    def __init__(self, channels: List[str], groups: List[str]) -> None:
         PNChannelGroupsEvent.__init__(self, channels, groups)
 
 
 class SubscriptionRestoredEvent(PNCursorEvent, PNChannelGroupsEvent):
-    def __init__(self, timetoken: str, region: int, channels: list[str], groups: list[str]) -> None:
+    def __init__(self, timetoken: str, region: int, channels: List[str], groups: List[str]) -> None:
         PNCursorEvent.__init__(self, timetoken, region)
         PNChannelGroupsEvent.__init__(self, channels, groups)
 

pubnub/event_engine/events.py

@@ -1,14 +1,19 @@
-import events
-import states
+from pubnub.event_engine import effects, events, states
+from pubnub.event_engine.dispatcher import Dispatcher
+from typing import List
 
 
 class StateMachine:
     _current_state: states.PNState
-    _context = states.PNContext()
+    _context: states.PNContext
+    _effect_list: List[effects.PNEffect]
 
     def __init__(self, initial_state: states.PNState) -> None:
+        self._context = states.PNContext()
         self._current_state = initial_state(self._context)
-        self._effect_queue = []
+        self._listeners = {}
+        self._effect_list = []
+        self._dispatcher = Dispatcher()
 
     def get_state_name(self):
         return self._current_state.__class__.__name__
@@ -17,28 +22,30 @@ def trigger(self, event: events.PNEvent) -> states.PNTransition:
         if event.get_name() in self._current_state._transitions:
             effect = self._current_state.on_exit()
             if effect:
-                self._effect_queue.append(effect)
+                self._effect_list.append(effect)
 
             transition: states.PNTransition = self._current_state.on(event, self._context)
 
             self._current_state = transition.state(self._current_state.get_context())
             self._context = transition.context
             if transition.effect:
-                self._effect_queue.append(transition.effect)
+                self._effect_list.append(transition.effect)
 
             effect = self._current_state.on_enter(self._context)
             if effect:
-                self._effect_queue.append(effect)
+                self._effect_list.append(effect)
 
             if transition.state:
                 self._current_state = transition.state(self._context)
 
         else:
             # we're ignoring events unhandled
             print('unhandled event??')
-            pass
 
-        return self._effect_queue
+        for effect in self._effect_list:
+            self._dispatcher.dispatch_effect(effect)
+
+        return self._effect_list
 
 
 if __name__ == "__main__":
@@ -48,6 +55,9 @@ def trigger(self, event: events.PNEvent) -> states.PNTransition:
         channels=['fail'], groups=[]
     ))
 
+    machine.add_listener(effects.PNEffect, lambda x: print(f'Catch All Logger: {effect.__dict__}'))
+
+    machine.add_listener(effects.EmitMessagesEffect, )
     effect = machine.trigger(events.DisconnectEvent())
     print(f'SubscriptionChangedEvent triggered with channels=[`fail`]. Current state: {machine.get_state_name()}')
-    print(f'effect queue: {machine._effect_queue}')
+    print(f'effect queue: {machine._effect_list}')

pubnub/event_engine/statemachine.py

@@ -1,11 +1,8 @@
-import events
-import effects
-
-from effects import PNEffect
-from typing import Union
-
 from pubnub.enums import PNStatusCategory
+from pubnub.event_engine import effects, events
+from pubnub.event_engine.effects import PNEffect
 from pubnub.exceptions import PubNubException
+from typing import List, Union
 
 
 class PNContext(dict):
@@ -25,7 +22,7 @@ class PNState:
 
     def __init__(self, context: PNContext) -> None:
         self._context = context
-        self._transitions = {}
+        self._transitions = dict
 
     def on(self, event: events.PNEvent, context: PNContext):
         return self._transitions[event.get_name()](event, context)
@@ -43,12 +40,12 @@ def get_context(self) -> PNContext:
 class PNTransition:
     context: PNContext
     state: PNState
-    effect: Union[None, list[PNEffect]]
+    effect: Union[None, List[PNEffect]]
 
     def __init__(self,
                  state: PNState,
                  context: Union[None, PNContext] = None,
-                 effect: Union[None, list[PNEffect]] = None,
+                 effect: Union[None, List[PNEffect]] = None,
                  ) -> None:
         self.context = context
         self.state = state
@@ -182,7 +179,7 @@ def on_enter(self, context: Union[None, PNContext]):
 
     def on_exit(self):
         super().on_exit()
-        return effects.CancelHandshakeEffect()
+        return effects.CancelHandshakeReconnectEffect()
 
     def disconnect(self, event: events.DisconnectEvent, context: PNContext) -> PNTransition:
         self._context.update(context)
@@ -411,7 +408,7 @@ def on_enter(self, context: Union[None, PNContext]):
 
     def on_exit(self):
         super().on_exit()
-        return effects.CancelReconnectEffect()
+        return effects.CancelReceiveReconnectEffect()
 
     def reconnect_failure(self, event: events.ReceiveReconnectFailureEvent, context: PNContext) -> PNTransition:
         self._context.update(context)

pubnub/event_engine/states.py

@@ -0,0 +1,17 @@
+from unittest.mock import patch
+from pubnub.event_engine import effects
+from pubnub.event_engine.dispatcher import Dispatcher
+
+
+def test_dispatch_emit_messages_effect():
+    with patch.object(effects.EmitEffect, 'emit_message') as mocked_emit_message:
+        dispatcher = Dispatcher()
+        dispatcher.dispatch_effect(effects.EmitMessagesEffect(['chan']))
+        mocked_emit_message.assert_called()
+
+
+def test_dispatch_emit_status_effect():
+    with patch.object(effects.EmitEffect, 'emit_status') as mocked_emit_status:
+        dispatcher = Dispatcher()
+        dispatcher.dispatch_effect(effects.EmitStatusEffect(['chan']))
+        mocked_emit_status.assert_called()