Repository: ably/ably-python (python)

Issue:
Set the `ATTACH_RESUME` flag on unclean attach
Spec: [`RTL4j`](https://sdk.ably.com/builds/ably/specification/main/features/#RTL4j)

* Add private `attach_resume` field to `RealtimeChannel` to track whether a channel has previously been attached
* Encode `ATTACH_RESUME` flag as per [`TR3f`](https://sdk.ably.com/builds/ably/specification/main/features/#TR3f)
* The testing method described in RTL4j2 won't work because we don't currently support channel rewind so we may need to come up with an alternative testing strategy
* Send the channel serial whenever the attach resume flag is used



Pull request description:
Implement attach resume, add test for receiving missed messages after disconnection
Resolves #385

Patch:
diff --git a/ably/realtime/connection.py b/ably/realtime/connection.py
index 5095a79..c5f78d0 100644
--- a/ably/realtime/connection.py
+++ b/ably/realtime/connection.py
@@ -413,9 +413,10 @@ class ConnectionManager(EventEmitter):
         except asyncio.CancelledError:
             return
 
-    def notify_state(self, state: ConnectionState, reason=None):
+    def notify_state(self, state: ConnectionState, reason=None, retry_immediately=None):
         # RTN15a
-        retry_immediately = state == ConnectionState.DISCONNECTED and self.__state == ConnectionState.CONNECTED
+        retry_immediately = (retry_immediately is not False) and (
+            state == ConnectionState.DISCONNECTED and self.__state == ConnectionState.CONNECTED)
 
         log.info(
             f'ConnectionManager.notify_state(): new state: {state}'
diff --git a/ably/realtime/realtime_channel.py b/ably/realtime/realtime_channel.py
index e53b4d5..885cf22 100644
--- a/ably/realtime/realtime_channel.py
+++ b/ably/realtime/realtime_channel.py
@@ -25,10 +25,29 @@ class ChannelState(str, Enum):
     FAILED = 'failed'
 
 
+class Flag(int, Enum):
+    # Channel attach state flags
+    HAS_PRESENCE = 1 << 0
+    HAS_BACKLOG = 1 << 1
+    RESUMED = 1 << 2
+    TRANSIENT = 1 << 4
+    ATTACH_RESUME = 1 << 5
+    # Channel mode flags
+    PRESENCE = 1 << 16
+    PUBLISH = 1 << 17
+    SUBSCRIBE = 1 << 18
+    PRESENCE_SUBSCRIBE = 1 << 19
+
+
+def has_flag(message_flags: int, flag: Flag):
+    return message_flags & flag > 0
+
+
 @dataclass
 class ChannelStateChange:
     previous: ChannelState
     current: ChannelState
+    resumed: bool
     reason: Optional[AblyException] = None
 
 
@@ -62,6 +81,8 @@ class RealtimeChannel(EventEmitter, Channel):
         self.__state = ChannelState.INITIALIZED
         self.__message_emitter = EventEmitter()
         self.__state_timer: Timer | None = None
+        self.__attach_resume = False
+        self.__channel_serial: str | None = None
 
         # Used to listen to state changes internally, if we use the public event emitter interface then internals
         # will be disrupted if the user called .off() to remove all listeners
@@ -116,6 +137,12 @@ class RealtimeChannel(EventEmitter, Channel):
             "action": ProtocolMessageAction.ATTACH,
             "channel": self.name,
         }
+
+        if self.__attach_resume:
+            attach_msg["flags"] = Flag.ATTACH_RESUME
+        if self.__channel_serial:
+            attach_msg["channelSerial"] = self.__channel_serial
+
         self._send_message(attach_msg)
 
     # RTL5
@@ -290,9 +317,17 @@ class RealtimeChannel(EventEmitter, Channel):
 
     def _on_message(self, msg):
         action = msg.get('action')
+
+        # RTL4c1
+        channel_serial = msg.get('channelSerial')
+        if channel_serial:
+            self.__channel_serial = channel_serial
+
         if action == ProtocolMessageAction.ATTACHED:
             if self.state == ChannelState.ATTACHING:
-                self._notify_state(ChannelState.ATTACHED)
+                flags = msg.get('flags')
+                resumed = has_flag(flags, Flag.RESUMED)
+                self._notify_state(ChannelState.ATTACHED, resumed=resumed)
             else:
                 log.warn("RealtimeChannel._on_message(): ATTACHED received while not attaching")
         elif action == ProtocolMessageAction.DETACHED:
@@ -310,7 +345,7 @@ class RealtimeChannel(EventEmitter, Channel):
         self._notify_state(state)
         self._check_pending_state()
 
-    def _notify_state(self, state: ChannelState, reason=None):
+    def _notify_state(self, state: ChannelState, reason=None, resumed=False):
         log.info(f'RealtimeChannel._notify_state(): state = {state}')
 
         self.__clear_state_timer()
@@ -318,7 +353,17 @@ class RealtimeChannel(EventEmitter, Channel):
         if state == self.state:
             return
 
-        state_change = ChannelStateChange(self.__state, state, reason=reason)
+        # RTL4j1
+        if state == ChannelState.ATTACHED:
+            self.__attach_resume = True
+        if state in (ChannelState.DETACHING, ChannelState.FAILED):
+            self.__attach_resume = False
+
+        # RTP5a1
+        if state in (ChannelState.DETACHED, ChannelState.SUSPENDED, ChannelState.FAILED):
+            self.__channel_serial = None
+
+        state_change = ChannelStateChange(self.__state, state, resumed, reason=reason)
 
         self.__state = state
         self._emit(state, state_change)


Test patch:
diff --git a/test/ably/realtimeresume_test.py b/test/ably/realtimeresume_test.py
index 7aba71f..58570f4 100644
--- a/test/ably/realtimeresume_test.py
+++ b/test/ably/realtimeresume_test.py
@@ -1,9 +1,24 @@
+import asyncio
 from ably.realtime.connection import ConnectionState
 from ably.realtime.realtime_channel import ChannelState
 from test.ably.restsetup import RestSetup
 from test.ably.utils import BaseAsyncTestCase, random_string
 
 
+async def send_and_await(rest_channel, realtime_channel):
+    event = random_string(5)
+    message = random_string(5)
+    future = asyncio.Future()
+
+    def on_message(_):
+        future.set_result(None)
+
+    await realtime_channel.subscribe(event, on_message)
+    await rest_channel.publish(event, message)
+
+    await future
+
+
 class TestRealtimeResume(BaseAsyncTestCase):
     async def asyncSetUp(self):
         self.test_vars = await RestSetup.get_test_vars()
@@ -108,3 +123,49 @@ class TestRealtimeResume(BaseAsyncTestCase):
         await channel.once_async(ChannelState.ATTACHED)
 
         await ably.close()
+
+    async def test_resume_receives_channel_messages_while_disconnected(self):
+        realtime = await RestSetup.get_ably_realtime()
+        rest = await RestSetup.get_ably_rest()
+
+        channel_name = random_string(5)
+
+        realtime_channel = realtime.channels.get(channel_name)
+        rest_channel = rest.channels.get(channel_name)
+
+        await realtime.connection.once_async(ConnectionState.CONNECTED)
+
+        asyncio.create_task(realtime_channel.attach())
+        state_change = await realtime_channel.once_async(ChannelState.ATTACHED)
+        assert state_change.resumed is False
+
+        await send_and_await(rest_channel, realtime_channel)
+
+        assert realtime.connection.connection_manager.transport
+        await realtime.connection.connection_manager.transport.dispose()
+        realtime.connection.connection_manager.notify_state(ConnectionState.DISCONNECTED, retry_immediately=False)
+
+        event_name = random_string(5)
+        message = random_string(5)
+        await rest_channel.publish(event_name, message)
+
+        future = asyncio.Future()
+
+        def on_message(message):
+            future.set_result(message)
+
+        await realtime_channel.subscribe(event_name, on_message)
+
+        realtime.connect()
+        await realtime.connection.once_async(ConnectionState.CONNECTED)
+
+        state_change = await realtime_channel.once_async(ChannelState.ATTACHED)
+
+        assert state_change.resumed is True
+
+        received_message = await future
+
+        assert received_message.data == message
+
+        await realtime.close()
+        await rest.close()