implement realtime presence publish/subscribe functionality

Author: owenpearson

ably/realtime/presencemap.py

@@ -89,6 +89,7 @@ def __init__(
         self._member_key_fn = member_key_fn
         self._is_newer_fn = is_newer_fn or _is_newer
         self._logger = logger_instance or logger
+        self._sync_complete_callbacks: List[Callable[[], None]] = []
 
     @property
     def sync_in_progress(self) -> bool:
@@ -303,8 +304,30 @@ def end_sync(self) -> Tuple[List[PresenceMessage], List[PresenceMessage]]:
                 f"{len(residual_list)} residual members"
             )
 
+            # Notify callbacks that sync is complete
+            for callback in self._sync_complete_callbacks:
+                try:
+                    callback()
+                except Exception as e:
+                    self._logger.error(f"Error in sync complete callback: {e}")
+            self._sync_complete_callbacks.clear()
+
         return residual_list, absent_list
 
+    def wait_sync(self, callback: Callable[[], None]) -> None:
+        """
+        Wait for SYNC to complete, calling callback when done.
+
+        If sync is not in progress, callback is called immediately.
+
+        Args:
+            callback: Function to call when sync completes
+        """
+        if not self._sync_in_progress:
+            callback()
+        else:
+            self._sync_complete_callbacks.append(callback)
+
     def clear(self) -> None:
         """
         Clear all members and reset sync state.
@@ -314,4 +337,5 @@ def clear(self) -> None:
         self._map.clear()
         self._residual_members = None
         self._sync_in_progress = False
+        self._sync_complete_callbacks.clear()
         self._logger.debug("PresenceMap.clear: cleared all members")

ably/realtime/realtime_channel.py

@@ -136,6 +136,10 @@ def __init__(self, realtime: AblyRealtime, name: str, channel_options: ChannelOp
         # will be disrupted if the user called .off() to remove all listeners
         self.__internal_state_emitter = EventEmitter()
 
+        # Initialize presence for this channel
+        from ably.realtime.realtimepresence import RealtimePresence
+        self.__presence = RealtimePresence(self)
+
         # Pass channel options as dictionary to parent Channel class
         Channel.__init__(self, realtime, name, self.__channel_options.to_dict())
 
@@ -528,6 +532,7 @@ def _on_message(self, proto_msg: dict) -> None:
             error = proto_msg.get("error")
             exception = None
             resumed = False
+            has_presence = False
 
             self.__attach_serial = channel_serial
             self.__channel_serial = channel_serial
@@ -538,14 +543,16 @@ def _on_message(self, proto_msg: dict) -> None:
 
             if flags:
                 resumed = has_flag(flags, Flag.RESUMED)
+                # RTP1: Check for HAS_PRESENCE flag
+                has_presence = has_flag(flags, Flag.HAS_PRESENCE)
 
             #  RTL12
             if self.state == ChannelState.ATTACHED:
                 if not resumed:
                     state_change = ChannelStateChange(self.state, ChannelState.ATTACHED, resumed, exception)
                     self._emit("update", state_change)
             elif self.state == ChannelState.ATTACHING:
-                self._notify_state(ChannelState.ATTACHED, resumed=resumed)
+                self._notify_state(ChannelState.ATTACHED, resumed=resumed, has_presence=has_presence)
             else:
                 log.warn("RealtimeChannel._on_message(): ATTACHED received while not attaching")
         elif action == ProtocolMessageAction.DETACHED:
@@ -568,6 +575,19 @@ def _on_message(self, proto_msg: dict) -> None:
                     log.error(f"Message processing error {e}. Skip messages {proto_msg.get('messages')}")
             for message in messages:
                 self.__message_emitter._emit(message.name, message)
+        elif action == ProtocolMessageAction.PRESENCE:
+            # Handle PRESENCE messages
+            from ably.types.presence import PresenceMessage
+            presence_messages = proto_msg.get('presence', [])
+            decoded_presence = PresenceMessage.from_encoded_array(presence_messages, cipher=self.cipher)
+            self.__presence.set_presence(decoded_presence, is_sync=False)
+        elif action == ProtocolMessageAction.SYNC:
+            # Handle SYNC messages (RTP18)
+            from ably.types.presence import PresenceMessage
+            presence_messages = proto_msg.get('presence', [])
+            decoded_presence = PresenceMessage.from_encoded_array(presence_messages, cipher=self.cipher)
+            sync_channel_serial = proto_msg.get('channelSerial')
+            self.__presence.set_presence(decoded_presence, is_sync=True, sync_channel_serial=sync_channel_serial)
         elif action == ProtocolMessageAction.ERROR:
             error = AblyException.from_dict(proto_msg.get('error'))
             self._notify_state(ChannelState.FAILED, reason=error)
@@ -578,7 +598,7 @@ def _request_state(self, state: ChannelState) -> None:
         self._check_pending_state()
 
     def _notify_state(self, state: ChannelState, reason: AblyException | None = None,
-                      resumed: bool = False) -> None:
+                      resumed: bool = False, has_presence: bool = False) -> None:
         log.debug(f'RealtimeChannel._notify_state(): state = {state}')
 
         self.__clear_state_timer()
@@ -616,6 +636,9 @@ def _notify_state(self, state: ChannelState, reason: AblyException | None = None
         self._emit(state, state_change)
         self.__internal_state_emitter._emit(state, state_change)
 
+        # RTP5: Notify presence of channel state change
+        self.__presence.act_on_channel_state(state, has_presence=has_presence, error=reason)
+
     def _send_message(self, msg: dict) -> None:
         asyncio.create_task(self.__realtime.connection.connection_manager.send_protocol_message(msg))
 
@@ -706,6 +729,11 @@ def params(self) -> dict[str, str]:
         """Get channel parameters"""
         return self.__params
 
+    @property
+    def presence(self):
+        """Get the RealtimePresence object for this channel"""
+        return self.__presence
+
     def _start_decode_failure_recovery(self, error: AblyException) -> None:
         """Start RTL18 decode failure recovery procedure"""
 

ably/transport/websockettransport.py

@@ -180,7 +180,9 @@ async def on_protocol_message(self, msg):
         elif action in (
             ProtocolMessageAction.ATTACHED,
             ProtocolMessageAction.DETACHED,
-            ProtocolMessageAction.MESSAGE
+            ProtocolMessageAction.MESSAGE,
+            ProtocolMessageAction.PRESENCE,
+            ProtocolMessageAction.SYNC
         ):
             self.connection_manager.on_channel_message(msg)