[AIT-96] feat: RealtimeChannel publish over WebSocket implementation

Implemented Spec points:

## Message Publishing Specifications (RTL6)

### RTL6c - Messages published on channels in specific states
- Messages published when channel is not **ATTACHED** should be published immediately

### RTL6c2 - Message queuing behavior
- Messages can be queued when connection/channel is not ready
- Relates to processing queued messages when connection becomes ready

### RTL6c3 - Publishing without implicit attach

### RTL6c4 - Behavior when queueMessages client option is false

### RTL6d - Message bundling restrictions

#### RTL6d1: Maximum message size limits for bundling
- **RTL6d2**: All messages in bundle must have same clientId

#### RTL6d3: Can only bundle messages for same channel
- **RTL6d4**: Can only bundle messages with same action (MESSAGE or PRESENCE)

#### RTL6d7: Cannot bundle idempotent messages with non-idempotent messages

---

## Message Acknowledgment (RTN7)

### RTN7a
All **PRESENCE**, **MESSAGE**, **ANNOTATION**, and **OBJECT** ProtocolMessages sent to Ably expect either an **ACK** or **NACK** to confirm successful receipt or failure

### RTN7b
Every ProtocolMessage requiring acknowledgment must contain a unique serially incrementing `msgSerial` integer starting at zero

### RTN7c
If connection enters **SUSPENDED**, **CLOSED**, or **FAILED** state and ACK/NACK has not been received, client should fail those messages and remove them from retry queues

### RTN7d
If `queueMessages` is false, messages entering **DISCONNECTED** state without acknowledgment should be treated as failed immediately

### RTN7e
When connection state changes to **SUSPENDED**/**CLOSED**/**FAILED**, pending messages (submitted via RTL6c1 or RTL6c2) awaiting ACK/NACK should be considered failed

---

## Message Resending and Serial Handling (RTN19)

### RTN19a
Upon reconnection after disconnection, client library must resend all pending messages awaiting acknowledgment, allowing the realtime system to respond with ACK/NACK

### RTN19a2
In the event of a new `connectionId` (connection not resumed), previous `msgSerials` are meaningless and must be reset. The `msgSerial` counter resets to 0 for the new connection

---

## Channel State and Reattachment (RTL3, RTL4, RTL5)

### RTL3c
Channel state implications when connection goes into **SUSPENDED**

### RTL3d
When connection enters **CONNECTED** state, channels in **ATTACHING**, **ATTACHED**, or **SUSPENDED** states should transition to **ATTACHING** and initiate attach sequence. Connection should process queued messages immediately without waiting for attach operations to finish

### RTL4c - Attach sequence
- **RTL4c1**: ATTACH message includes channel serial to resume from previous message or attachment

### RTL5i
If channel is **DETACHING**, re-send **DETACH** and remain in 'detaching' state

Author: ttypic

ably/realtime/connectionmanager.py

@@ -24,6 +24,85 @@
 log = logging.getLogger(__name__)
 
 
+class PendingMessage:
+    """Represents a message awaiting acknowledgment from the server"""
+
+    def __init__(self, message: dict):
+        self.message = message
+        self.future: asyncio.Future | None = None
+        action = message.get('action')
+
+        # Messages that require acknowledgment: MESSAGE, PRESENCE, ANNOTATION, OBJECT
+        self.ack_required = action in (
+            ProtocolMessageAction.MESSAGE,
+            ProtocolMessageAction.PRESENCE,
+            ProtocolMessageAction.ANNOTATION,
+            ProtocolMessageAction.OBJECT,
+        )
+
+
+class PendingMessageQueue:
+    """Queue for tracking messages awaiting acknowledgment"""
+
+    def __init__(self):
+        self.messages: list[PendingMessage] = []
+
+    def push(self, pending_message: PendingMessage) -> None:
+        """Add a message to the queue"""
+        self.messages.append(pending_message)
+
+    def count(self) -> int:
+        """Return the number of pending messages"""
+        return len(self.messages)
+
+    def complete_messages(self, serial: int, count: int, err: AblyException | None = None) -> None:
+        """Complete messages based on serial and count from ACK/NACK
+
+        Args:
+            serial: The msgSerial of the first message being acknowledged
+            count: The number of messages being acknowledged
+            err: Error from NACK, or None for successful ACK
+        """
+        log.debug(f'MessageQueue.complete_messages(): serial={serial}, count={count}, err={err}')
+
+        if not self.messages:
+            log.warning('MessageQueue.complete_messages(): called on empty queue')
+            return
+
+        first = self.messages[0]
+        if first:
+            start_serial = first.message.get('msgSerial')
+            if start_serial is None:
+                log.warning('MessageQueue.complete_messages(): first message has no msgSerial')
+                return
+
+            end_serial = serial + count
+
+            if end_serial > start_serial:
+                # Remove and complete the acknowledged messages
+                num_to_complete = min(end_serial - start_serial, len(self.messages))
+                completed_messages = self.messages[:num_to_complete]
+                self.messages = self.messages[num_to_complete:]
+
+                for msg in completed_messages:
+                    if msg.future and not msg.future.done():
+                        if err:
+                            msg.future.set_exception(err)
+                        else:
+                            msg.future.set_result(None)
+
+    def complete_all_messages(self, err: AblyException) -> None:
+        """Complete all pending messages with an error"""
+        while self.messages:
+            msg = self.messages.pop(0)
+            if msg.future and not msg.future.done():
+                msg.future.set_exception(err)
+
+    def clear(self) -> None:
+        """Clear all messages from the queue"""
+        self.messages.clear()
+
+
 class ConnectionManager(EventEmitter):
     def __init__(self, realtime: AblyRealtime, initial_state):
         self.options = realtime.options
@@ -43,6 +122,8 @@ def __init__(self, realtime: AblyRealtime, initial_state):
         self.__fallback_hosts: list[str] = self.options.get_fallback_realtime_hosts()
         self.queued_messages: Queue = Queue()
         self.__error_reason: AblyException | None = None
+        self.msg_serial: int = 0
+        self.pending_message_queue: PendingMessageQueue = PendingMessageQueue()
         super().__init__()
 
     def enact_state_change(self, state: ConnectionState, reason: AblyException | None = None) -> None:
@@ -87,38 +168,131 @@ async def close_impl(self) -> None:
 
         self.notify_state(ConnectionState.CLOSED)
 
-    async def send_protocol_message(self, protocol_message: dict) -> None:
+    async def send_protocol_message(
+        self, message_or_pending: dict | PendingMessage
+    ) -> None:
+        """Send a protocol message and optionally track it for acknowledgment
+
+        Args:
+            message_or_pending: Either a protocol message dict (new message) or a PendingMessage (requeued message)
+        Returns:
+            None
+        """
+        # Handle both new messages (dict) and requeued messages (PendingMessage)
+        if isinstance(message_or_pending, PendingMessage):
+            # Requeued message - reuse existing PendingMessage with its Future
+            pending_message = message_or_pending
+            protocol_message = pending_message.message
+        else:
+            # New message - create PendingMessage wrapper
+            protocol_message = message_or_pending
+            pending_message = PendingMessage(protocol_message)
+
+            # Create a future for async/await support
+            if pending_message.ack_required:
+                pending_message.future = asyncio.Future()
+
+            # Assign msgSerial to messages that need acknowledgment
+            if pending_message.ack_required:
+                # New message - assign fresh serial
+                protocol_message['msgSerial'] = self.msg_serial
+                self.msg_serial += 1
+
         if self.state in (
             ConnectionState.DISCONNECTED,
             ConnectionState.CONNECTING,
         ):
-            self.queued_messages.put(protocol_message)
-            return
+            self.queued_messages.put(pending_message)
+            # For queued messages requiring ack, add to pending queue
+            if pending_message.ack_required:
+                self.pending_message_queue.push(pending_message)
+
+            if pending_message.ack_required:
+                await pending_message.future
+            return None
 
         if self.state == ConnectionState.CONNECTED:
             if self.transport:
+                # Add to pending queue before sending
+                if pending_message.ack_required:
+                    self.pending_message_queue.push(pending_message)
+
                 await self.transport.send(protocol_message)
             else:
                 log.exception(
                     "ConnectionManager.send_protocol_message(): can not send message with no active transport"
                 )
-            return
-
-        raise AblyException(f"ConnectionManager.send_protocol_message(): called in {self.state}", 500, 50000)
+                if pending_message.future:
+                    pending_message.future.set_exception(
+                        AblyException("No active transport", 500, 50000)
+                    )
+            if pending_message.ack_required:
+                await pending_message.future
+            return None
+
+        error = AblyException(f"ConnectionManager.send_protocol_message(): called in {self.state}", 500, 50000)
+        if pending_message.future:
+            pending_message.future.set_exception(error)
+        raise error
 
     def send_queued_messages(self) -> None:
         log.info(f'ConnectionManager.send_queued_messages(): sending {self.queued_messages.qsize()} message(s)')
         while not self.queued_messages.empty():
-            asyncio.create_task(self.send_protocol_message(self.queued_messages.get()))
+            pending_message = self.queued_messages.get()
+            asyncio.create_task(self.send_protocol_message(pending_message))
+
+    def requeue_pending_messages(self) -> None:
+        """RTN19a: Requeue messages awaiting ACK/NACK when transport disconnects
+
+        These messages will be resent when connection becomes CONNECTED again.
+        RTN19a2: msgSerial is preserved for resume, reset for new connection.
+        """
+        pending_count = self.pending_message_queue.count()
+        if pending_count == 0:
+            return
+
+        log.info(
+            f'ConnectionManager.requeue_pending_messages(): '
+            f'requeuing {pending_count} pending message(s) for resend'
+        )
+
+        # Get all pending messages and add them back to the queue
+        # They'll be sent again when we reconnect
+        pending_messages = list(self.pending_message_queue.messages)
+
+        # Add back to front of queue (FIFO but priority over new messages)
+        # Store the entire PendingMessage object to preserve Future
+        for pending_msg in reversed(pending_messages):
+            # PendingMessage object retains its Future, msgSerial
+            self.queued_messages.put(pending_msg)
+
+        # Clear the message queue since we're requeueing them all
+        # When they're resent, the existing Future will be resolved
+        self.pending_message_queue.clear()
 
     def fail_queued_messages(self, err) -> None:
         log.info(
             f"ConnectionManager.fail_queued_messages(): discarding {self.queued_messages.qsize()} messages;" +
             f" reason = {err}"
         )
+        error = err or AblyException("Connection failed", 80000, 500)
         while not self.queued_messages.empty():
-            msg = self.queued_messages.get()
-            log.exception(f"ConnectionManager.fail_queued_messages(): Failed to send protocol message: {msg}")
+            pending_msg = self.queued_messages.get()
+            log.exception(
+                f"ConnectionManager.fail_queued_messages(): Failed to send protocol message: "
+                f"{pending_msg.message}"
+            )
+            # Fail the Future if it exists
+            if pending_msg.future and not pending_msg.future.done():
+                pending_msg.future.set_exception(error)
+
+        # Also fail all pending messages awaiting acknowledgment
+        if self.pending_message_queue.count() > 0:
+            count = self.pending_message_queue.count()
+            log.info(
+                f"ConnectionManager.fail_queued_messages(): failing {count} pending messages"
+            )
+            self.pending_message_queue.complete_all_messages(error)
 
     async def ping(self) -> float:
         if self.__ping_future:
@@ -149,6 +323,16 @@ def on_connected(self, connection_details: ConnectionDetails, connection_id: str
                      reason: AblyException | None = None) -> None:
         self.__fail_state = ConnectionState.DISCONNECTED
 
+        # RTN19a2: Reset msgSerial if connectionId changed (new connection)
+        prev_connection_id = self.connection_id
+        connection_id_changed = prev_connection_id is not None and prev_connection_id != connection_id
+
+        if connection_id_changed:
+            log.info('ConnectionManager.on_connected(): New connectionId; resetting msgSerial')
+            self.msg_serial = 0
+            # Note: In JS they call resetSendAttempted() here, but we don't need it
+            # because we fail all pending messages on disconnect per RTN7e
+
         self.__connection_details = connection_details
         self.connection_id = connection_id
 
@@ -244,7 +428,36 @@ def on_heartbeat(self, id: str | None) -> None:
                     self.__ping_future.set_result(None)
                 self.__ping_future = None
 
+    def on_ack(self, serial: int, count: int) -> None:
+        """Handle ACK protocol message from server
+
+        Args:
+            serial: The msgSerial of the first message being acknowledged
+            count: The number of messages being acknowledged
+        """
+        log.debug(f'ConnectionManager.on_ack(): serial={serial}, count={count}')
+        self.pending_message_queue.complete_messages(serial, count)
+
+    def on_nack(self, serial: int, count: int, err: AblyException | None) -> None:
+        """Handle NACK protocol message from server
+
+        Args:
+            serial: The msgSerial of the first message being rejected
+            count: The number of messages being rejected
+            err: Error information from the server
+        """
+        if not err:
+            err = AblyException('Unable to send message; channel not responding', 50001, 500)
+
+        log.error(f'ConnectionManager.on_nack(): serial={serial}, count={count}, err={err}')
+        self.pending_message_queue.complete_messages(serial, count, err)
+
     def deactivate_transport(self, reason: AblyException | None = None):
+        # RTN19a: Before disconnecting, requeue any pending messages
+        # so they'll be resent on reconnection
+        if self.transport:
+            log.info('ConnectionManager.deactivate_transport(): requeuing pending messages')
+            self.requeue_pending_messages()
         self.transport = None
         self.notify_state(ConnectionState.DISCONNECTED, reason)
 
@@ -383,8 +596,16 @@ def notify_state(self, state: ConnectionState, reason: AblyException | None = No
             ConnectionState.SUSPENDED,
             ConnectionState.FAILED,
         ):
+            # RTN7e: Fail pending messages on SUSPENDED, CLOSED, FAILED
             self.fail_queued_messages(reason)
             self.ably.channels._propagate_connection_interruption(state, reason)
+        elif state == ConnectionState.DISCONNECTED and not self.options.queue_messages:
+            # RTN7d: If queueMessages is false, fail pending messages on DISCONNECTED
+            log.info(
+                'ConnectionManager.notify_state(): queueMessages is false; '
+                'failing pending messages on DISCONNECTED'
+            )
+            self.fail_queued_messages(reason)
 
     def start_transition_timer(self, state: ConnectionState, fail_state: ConnectionState | None = None) -> None:
         log.debug(f'ConnectionManager.start_transition_timer(): transition state = {state}')
@@ -466,6 +687,8 @@ def cancel_retry_timer(self) -> None:
     def disconnect_transport(self) -> None:
         log.info('ConnectionManager.disconnect_transport()')
         if self.transport:
+            # RTN19a: Requeue pending messages before disposing transport
+            self.requeue_pending_messages()
             self.disconnect_transport_task = asyncio.create_task(self.transport.dispose())
 
     async def on_auth_updated(self, token_details: TokenDetails):