Presence engine - states and events

pubnub/event_engine/manage_effects.py

@@ -3,6 +3,7 @@
 import math
 
 from typing import Optional, Union
+from pubnub.endpoints.presence.heartbeat import Heartbeat
 from pubnub.endpoints.pubsub.subscribe import Subscribe
 from pubnub.enums import PNReconnectionPolicy
 from pubnub.exceptions import PubNubException
@@ -313,3 +314,61 @@ def emit_status(self, effect: effects.EmitStatusEffect):
         pn_status.category = effect.status
         pn_status.error = False
         self.pubnub._subscription_manager._listener_manager.announce_status(pn_status)
+
+
+class ManageHeartbeatEffect(ManagedEffect):
+    def run(self):
+        channels = self.effect.channels
+        groups = self.effect.groups
+        if hasattr(self.pubnub, 'event_loop'):
+            self.stop_event = self.get_new_stop_event()
+
+            loop: asyncio.AbstractEventLoop = self.pubnub.event_loop
+            coro = self.heartbeat(channels=channels, groups=groups, stop_event=self.stop_event)
+            if loop.is_running():
+                loop.create_task(coro)
+            else:
+                loop.run_until_complete(coro)
+
+    async def heartbeat(self, channels, groups, stop_event):
+        request = Heartbeat(self.pubnub).channels(channels).channel_groups(groups).cancellation_event(stop_event)
+        heartbeat = await request.future()
+
+        if heartbeat.status.error:
+            self.logger.warning(f'Heartbeat failed: {heartbeat.status.error_data.__dict__}')
+            self.event_engine.trigger(events.HeartbeatFailureEvent(heartbeat.status.error_data, 1))
+        else:
+            self.event_engine.trigger(events.HeartbeatSuccessEvent(channels=channels, groups=groups))
+
+
+class ManageHeartbeatWaitEffect(ManagedEffect):
+    def __init__(self, pubnub_instance, event_engine_instance,
+                 effect: Union[effects.PNManageableEffect, effects.PNCancelEffect]) -> None:
+        super().__init__(pubnub_instance, event_engine_instance, effect)
+        self.heartbeat_interval = pubnub_instance.config.heartbeat_interval
+
+    def run(self):
+        if hasattr(self.pubnub, 'event_loop'):
+            self.stop_event = self.get_new_stop_event()
+            loop: asyncio.AbstractEventLoop = self.pubnub.event_loop
+            coro = self.heartbeat_wait(self.heartbeat_interval, stop_event=self.stop_event)
+            if loop.is_running():
+                loop.create_task(coro)
+            else:
+                loop.run_until_complete(coro)
+
+    async def heartbeat(self, wait_time: int, stop_event):
+        try:
+            await asyncio.sleep(wait_time)
+            self.event_engine.trigger(events.HeartbeatTimesUpEvent(channels=self.effect.channels,
+                                                                   groups=self.effect.groups))
+        except asyncio.CancelledError:
+            pass
+
+
+class ManageHeartbeatLeaveEffect(ManagedEffect):
+    pass
+
+
+class ManageHeartbeatDelayedHeartbeatEffect(ManagedEffect):
+    pass

pubnub/event_engine/models/effects.py

@@ -93,3 +93,48 @@ class EmitStatusEffect(PNEmittableEffect):
     def __init__(self, status: Union[None, PNStatusCategory]) -> None:
         super().__init__()
         self.status = status
+
+
+"""
+    Presence Effects
+"""
+
+
+class HeartbeatEffect(PNEffect):
+    def __init__(self) -> None:
+        super().__init__()
+
+
+class HeartbeatWaitEffect(PNEffect):
+    def __init__(self) -> None:
+        super().__init__()
+
+
+class HeartbeatCancelWaitEffect(PNEffect):
+    def __init__(self) -> None:
+        super().__init__()
+
+
+class HeartbeatLeaveEffect(PNEffect):
+    def __init__(self) -> None:
+        super().__init__()
+
+
+class HeartbeatWaitEffect(PNEffect):
+    def __init__(self) -> None:
+        super().__init__()
+
+
+class HeartbeatCancelWaitEffect(PNEffect):
+    def __init__(self) -> None:
+        super().__init__()
+
+
+class HeartbeatDelayedEffect(PNEffect):
+    def __init__(self) -> None:
+        super().__init__()
+
+
+class HeartbeatCancelDelayedEffect(PNEffect):
+    def __init__(self) -> None:
+        super().__init__()

pubnub/event_engine/models/events.py

@@ -97,3 +97,47 @@ class DisconnectEvent(PNEvent):
 
 class ReconnectEvent(PNEvent):
     pass
+
+
+"""
+    Presence Events
+"""
+
+
+class HeartbeatJoinedEvent(PNChannelGroupsEvent):
+    pass
+
+
+class HeartbeatReconnectEvent(PNChannelGroupsEvent, PNFailureEvent):
+    def __init__(self, channels: List[str], groups: List[str], reason: PubNubException, attempt: int,
+                 timetoken: int = 0) -> None:
+        PNChannelGroupsEvent.__init__(self, channels, groups)
+        PNFailureEvent.__init__(self, reason, attempt, timetoken)
+
+
+class HeartbeatLeftAllEvent(PNEvent):
+    pass
+
+
+class HeartbeatLeftEvent(PNChannelGroupsEvent):
+    pass
+
+
+class HeartbeatDisconnectEvent(PNChannelGroupsEvent):
+    pass
+
+
+class HeartbeatSuccessEvent(PNChannelGroupsEvent):
+    pass
+
+
+class HeartbeatFailureEvent(PNChannelGroupsEvent, PNFailureEvent):
+    pass
+
+
+class HeartbeatTimesUpEvent(PNChannelGroupsEvent, PNFailureEvent):
+    pass
+
+
+class HeartbeatGiveUpEvent(PNChannelGroupsEvent, PNFailureEvent):
+    pass