Repository: ably/ably-python (python)

Issue:
Add feature manifest to `integration/realtime`
Adding a feature manifest for the ongoing realtime client work will allow us to easily track which features are implemented by the realtime client.
File name/location should follow the convention described [here](https://github.com/ably/features#future-direction-for-the-sdk-manifests)

Pull request description:
Realtime feature manifest
Resolves #403 

Patch:
diff --git a/.ably/capabilities.yaml b/.ably/capabilities.yaml
new file mode 100644
index 0000000..965c2e7
--- /dev/null
+++ b/.ably/capabilities.yaml
@@ -0,0 +1,88 @@
+%YAML 1.2
+---
+common-version: 1.2.0-alpha.1
+compliance:
+  Agent Identifier:
+    Agents:
+  Authentication:
+    API Key:
+    Token:
+      Callback:
+      Literal:
+      URL:
+        Query Time:
+  Debugging:
+    Error Information:
+    Logs:
+  Protocol:
+    JSON:
+    MessagePack:
+      .caveats: 'Not supported for realtime'
+  Realtime:
+    Channel:
+      Attach:
+      Subscribe:
+      State Events:
+    Connection:
+      Disconnected Retry Timeout:
+      Lifecycle control:
+      Ping:
+      State Events:
+      Suspended Retry Timeout:
+  REST:
+    Authentication:
+      Authorize:
+      Create Token Request:
+      Get Client Identifier:
+      Request Token:
+    Channel:
+      Encryption:
+      Existence Check:
+      Get:
+      History:
+      Iterate:
+      Name:
+      Presence:
+        History:
+        Member List:
+      Publish:
+        Idempotence:
+      Push Notifications:
+        List Subscriptions:
+        Subscribe:
+      Release:
+      Status:
+        Channel Details:
+    Opaque Request:
+    Push Notifications Administration:
+      Channel Subscription:
+        List:
+        List Channels:
+        Remove:
+        Save:
+      Device Registration:
+        Get:
+        List:
+        Remove:
+        Save:
+      Publish:
+    Request Timeout:
+    Service:
+      Get Time:
+    Statistics:
+      Query:
+  Service:
+    Environment:
+    Fallbacks:
+      Hosts:
+      Retry Count:
+      Retry Duration:
+      Retry Timeout:
+    Host:
+  Testing:
+    Disable TLS:
+    TCP Insecure Port:
+    TCP Secure Port:
+  Transport:
+    Connection Open Timeout:
+    HTTP/2:
diff --git a/ably/realtime/connection.py b/ably/realtime/connection.py
index 951eda0..b26ab9b 100644
--- a/ably/realtime/connection.py
+++ b/ably/realtime/connection.py
@@ -4,14 +4,13 @@ import asyncio
 import httpx
 from ably.realtime.websockettransport import WebSocketTransport, ProtocolMessageAction
 from ably.transport.defaults import Defaults
-from ably.util.exceptions import AblyException
+from ably.util.exceptions import AblyAuthException, AblyException
 from ably.util.eventemitter import EventEmitter
 from enum import Enum
 from datetime import datetime
 from ably.util import helper
 from dataclasses import dataclass
 from typing import Optional
-from ably.types.connectiondetails import ConnectionDetails
 
 log = logging.getLogger(__name__)
 
@@ -47,6 +46,18 @@ class ConnectionStateChange:
     reason: Optional[AblyException] = None
 
 
+@dataclass
+class ConnectionDetails:
+    connectionStateTtl: int
+
+    def __init__(self, connection_state_ttl: int):
+        self.connectionStateTtl = connection_state_ttl
+
+    @staticmethod
+    def from_dict(json_dict: dict):
+        return ConnectionDetails(json_dict.get('connectionStateTtl'))
+
+
 class Connection(EventEmitter):
     """Ably Realtime Connection
 
@@ -155,10 +166,9 @@ class ConnectionManager(EventEmitter):
         self.__closed_future = None
         self.__ping_future = None
         self.__timeout_in_secs = self.options.realtime_request_timeout / 1000
-        self.retry_connection_attempt_task = None
-        self.connection_attempt_task = None
         self.transport: WebSocketTransport | None = None
         self.__ttl_task = None
+        self.__retry_task = None
         self.__connection_details = None
         self.__fail_state = ConnectionState.DISCONNECTED
         super().__init__()
@@ -173,12 +183,15 @@ class ConnectionManager(EventEmitter):
 
     async def __start_suspended_timer(self):
         if self.__connection_details:
-            self.ably.options.connection_state_ttl = self.__connection_details.connection_state_ttl
+            self.ably.options.connection_state_ttl = self.__connection_details.connectionStateTtl
         await asyncio.sleep(self.ably.options.connection_state_ttl / 1000)
         exception = AblyException("Exceeded connectionStateTtl while in DISCONNECTED state", 504, 50003)
         self.enact_state_change(ConnectionState.SUSPENDED, exception)
         self.__connection_details = None
         self.__fail_state = ConnectionState.SUSPENDED
+        if self.__retry_task:
+            self.__retry_task.cancel()
+            self.__retry_task = asyncio.create_task(self.retry_connection_attempt())
 
     async def connect(self):
         if not self.__connected_future:
@@ -187,8 +200,8 @@ class ConnectionManager(EventEmitter):
         await self.__connected_future
 
     def try_connect(self):
-        self.connection_attempt_task = asyncio.create_task(self._connect())
-        self.connection_attempt_task.add_done_callback(self.on_connection_attempt_done)
+        task = asyncio.create_task(self._connect())
+        task.add_done_callback(self.on_connection_attempt_done)
 
     async def _connect(self):
         if self.__state == ConnectionState.CONNECTED:
@@ -217,14 +230,6 @@ class ConnectionManager(EventEmitter):
             return False
 
     def on_connection_attempt_done(self, task):
-        if self.connection_attempt_task:
-            if not self.connection_attempt_task.done():
-                self.connection_attempt_task.cancel()
-            self.connection_attempt_task = None
-        if self.retry_connection_attempt_task:
-            if not self.retry_connection_attempt_task.done():
-                self.retry_connection_attempt_task.cancel()
-            self.retry_connection_attempt_task = None
         try:
             exception = task.exception()
         except asyncio.CancelledError:
@@ -238,8 +243,8 @@ class ConnectionManager(EventEmitter):
             if self.__connected_future:
                 self.__connected_future.set_exception(exception)
                 self.__connected_future = None
-            self.enact_state_change(ConnectionState.DISCONNECTED, exception)
-        self.retry_connection_attempt_task = asyncio.create_task(self.retry_connection_attempt())
+            self.enact_state_change(self.__fail_state, exception)
+        self.__retry_task = asyncio.create_task(self.retry_connection_attempt())
 
     async def retry_connection_attempt(self):
         if self.__fail_state == ConnectionState.SUSPENDED:
@@ -288,7 +293,6 @@ class ConnectionManager(EventEmitter):
 
     async def connect_impl(self):
         self.transport = WebSocketTransport(self)
-        self._emit('transport.pending', self.transport)
         await self.transport.connect()
         try:
             await asyncio.wait_for(asyncio.shield(self.__connected_future), self.__timeout_in_secs)
@@ -333,54 +337,56 @@ class ConnectionManager(EventEmitter):
         response_time_ms = (ping_end_time - ping_start_time) * 1000
         return round(response_time_ms, 2)
 
-    def on_connected(self, connection_details: ConnectionDetails):
-        if self.transport:
-            self.transport.is_connected = True
-        if self.__connected_future:
-            if not self.__connected_future.cancelled():
-                self.__connected_future.set_result(None)
-            self.__connected_future = None
-        self.__fail_state = ConnectionState.DISCONNECTED
-        if self.__ttl_task:
-            self.__ttl_task.cancel()
-        self.__connection_details = connection_details
-        if self.__state == ConnectionState.CONNECTED:
-            state_change = ConnectionStateChange(ConnectionState.CONNECTED, ConnectionState.CONNECTED,
-                                                 ConnectionEvent.UPDATE)
-            self._emit(ConnectionEvent.UPDATE, state_change)
-        else:
-            self.enact_state_change(ConnectionState.CONNECTED)
-
-    async def on_error(self, msg: dict, exception: AblyException):
-        if msg.get('channel') is None:
-            self.enact_state_change(ConnectionState.FAILED, exception)
+    async def on_protocol_message(self, msg):
+        action = msg['action']
+        if action == ProtocolMessageAction.CONNECTED:  # CONNECTED
+            if self.transport:
+                self.transport.is_connected = True
             if self.__connected_future:
-                self.__connected_future.set_exception(exception)
+                if not self.__connected_future.cancelled():
+                    self.__connected_future.set_result(None)
                 self.__connected_future = None
+            else:
+                log.warn('CONNECTED message received but connected_future not set')
+            self.__fail_state == ConnectionState.DISCONNECTED
+            if self.__ttl_task:
+                self.__ttl_task.cancel()
+            self.__connection_details = ConnectionDetails.from_dict(msg["connectionDetails"])
+            if self.__state == ConnectionState.CONNECTED:
+                state_change = ConnectionStateChange(ConnectionState.CONNECTED, ConnectionState.CONNECTED,
+                                                     ConnectionEvent.UPDATE)
+                self._emit(ConnectionEvent.UPDATE, state_change)
+            else:
+                self.enact_state_change(ConnectionState.CONNECTED)
+        if action == ProtocolMessageAction.ERROR:  # ERROR
+            error = msg["error"]
+            if error['nonfatal'] is False:
+                exception = AblyAuthException(error["message"], error["statusCode"], error["code"])
+                self.enact_state_change(ConnectionState.FAILED, exception)
+                if self.__connected_future:
+                    self.__connected_future.set_exception(exception)
+                    self.__connected_future = None
+                if self.transport:
+                    await self.transport.dispose()
+                raise exception
+        if action == ProtocolMessageAction.CLOSED:
             if self.transport:
                 await self.transport.dispose()
-            raise exception
-
-    async def on_closed(self):
-        if self.transport:
-            await self.transport.dispose()
-        if self.__closed_future and not self.__closed_future.done():
             self.__closed_future.set_result(None)
-
-    def on_channel_message(self, msg: dict):
-        self.__ably.channels._on_channel_message(msg)
-
-    def on_heartbeat(self, id: Optional[str]):
-        if self.__ping_future:
-            # Resolve on heartbeat from ping request.
-            if self.__ping_id == id:
-                if not self.__ping_future.cancelled():
-                    self.__ping_future.set_result(None)
-                self.__ping_future = None
-
-    def deactivate_transport(self, reason=None):
-        self.transport = None
-        self.enact_state_change(ConnectionState.DISCONNECTED, reason)
+        if action == ProtocolMessageAction.HEARTBEAT:
+            if self.__ping_future:
+                # Resolve on heartbeat from ping request.
+                # TODO: Handle Normal heartbeat if required
+                if self.__ping_id == msg.get("id"):
+                    if not self.__ping_future.cancelled():
+                        self.__ping_future.set_result(None)
+                    self.__ping_future = None
+        if action in (
+            ProtocolMessageAction.ATTACHED,
+            ProtocolMessageAction.DETACHED,
+            ProtocolMessageAction.MESSAGE
+        ):
+            self.__ably.channels._on_channel_message(msg)
 
     @property
     def ably(self):
diff --git a/ably/realtime/websockettransport.py b/ably/realtime/websockettransport.py
index 513a2ac..3aeafcc 100644
--- a/ably/realtime/websockettransport.py
+++ b/ably/realtime/websockettransport.py
@@ -8,9 +8,7 @@ import socket
 import urllib.parse
 from ably.http.httputils import HttpUtils
 from ably.transport.defaults import Defaults
-from ably.types.connectiondetails import ConnectionDetails
 from ably.util.exceptions import AblyException
-from ably.util.helper import Timer, unix_time_ms
 from websockets.client import WebSocketClientProtocol, connect as ws_connect
 from websockets.exceptions import ConnectionClosedOK, WebSocketException
 
@@ -40,11 +38,7 @@ class WebSocketTransport:
         self.connect_task: asyncio.Task | None = None
         self.ws_connect_task: asyncio.Task | None = None
         self.connection_manager = connection_manager
-        self.options = self.connection_manager.options
         self.is_connected = False
-        self.idle_timer = None
-        self.last_activity = None
-        self.max_idle_interval = None
 
     async def connect(self):
         headers = HttpUtils.default_headers()
@@ -78,35 +72,6 @@ class WebSocketTransport:
         except (WebSocketException, socket.gaierror) as e:
             raise AblyException(f'Error opening websocket connection: {e}', 400, 40000)
 
-    async def on_protocol_message(self, msg):
-        self.on_activity()
-        log.info(f'WebSocketTransport.on_protocol_message(): receieved protocol message: {msg}')
-        action = msg.get('action')
-        if action == ProtocolMessageAction.CONNECTED:
-            connection_details = ConnectionDetails.from_dict(msg.get('connectionDetails'))
-            max_idle_interval = connection_details.max_idle_interval
-            if max_idle_interval:
-                self.max_idle_interval = max_idle_interval + self.options.realtime_request_timeout
-                self.on_activity()
-            self.connection_manager.on_connected(connection_details)
-        elif action == ProtocolMessageAction.CLOSED:
-            if self.ws_connect_task:
-                self.ws_connect_task.cancel()
-            await self.connection_manager.on_closed()
-        elif action == ProtocolMessageAction.ERROR:
-            error = msg.get('error')
-            exception = AblyException(error.get('message'), error.get('statusCode'), error.get('code'))
-            await self.connection_manager.on_error(msg, exception)
-        elif action == ProtocolMessageAction.HEARTBEAT:
-            id = msg.get('id')
-            self.connection_manager.on_heartbeat(id)
-        elif action in (
-            ProtocolMessageAction.ATTACHED,
-            ProtocolMessageAction.DETACHED,
-            ProtocolMessageAction.MESSAGE
-        ):
-            self.connection_manager.on_channel_message(msg)
-
     async def ws_read_loop(self):
         while True:
             if self.websocket is not None:
@@ -115,7 +80,11 @@ class WebSocketTransport:
                 except ConnectionClosedOK:
                     break
                 msg = json.loads(raw)
-                await self.on_protocol_message(msg)
+                log.info(f'ws_read_loop(): receieved protocol message: {msg}')
+                if msg['action'] == ProtocolMessageAction.CLOSED:
+                    if self.ws_connect_task:
+                        self.ws_connect_task.cancel()
+                await self.connection_manager.on_protocol_message(msg)
             else:
                 raise Exception('ws_read_loop running with no websocket')
 
@@ -132,8 +101,6 @@ class WebSocketTransport:
             self.read_loop.cancel()
         if self.ws_connect_task:
             self.ws_connect_task.cancel()
-        if self.idle_timer:
-            self.idle_timer.cancel()
         if self.websocket:
             try:
                 await self.websocket.close()
@@ -149,28 +116,3 @@ class WebSocketTransport:
         raw_msg = json.dumps(message)
         log.info(f'WebSocketTransport.send(): sending {raw_msg}')
         await self.websocket.send(raw_msg)
-
-    def set_idle_timer(self, timeout: float):
-        if not self.idle_timer:
-            self.idle_timer = Timer(timeout, self.on_idle_timer_expire)
-
-    async def on_idle_timer_expire(self):
-        self.idle_timer = None
-        since_last = unix_time_ms() - self.last_activity
-        time_remaining = self.max_idle_interval - since_last
-        msg = f"No activity seen from realtime in {since_last} ms; assuming connection has dropped"
-        if time_remaining <= 0:
-            log.error(msg)
-            await self.disconnect(AblyException(msg, 408, 80003))
-        else:
-            self.set_idle_timer(time_remaining + 100)
-
-    def on_activity(self):
-        if not self.max_idle_interval:
-            return
-        self.last_activity = unix_time_ms()
-        self.set_idle_timer(self.max_idle_interval + 100)
-
-    async def disconnect(self, reason=None):
-        await self.dispose()
-        self.connection_manager.deactivate_transport(reason)
diff --git a/ably/types/connectiondetails.py b/ably/types/connectiondetails.py
deleted file mode 100644
index c338f6e..0000000
--- a/ably/types/connectiondetails.py
+++ /dev/null
@@ -1,15 +0,0 @@
-from dataclasses import dataclass
-
-
-@dataclass()
-class ConnectionDetails:
-    connection_state_ttl: int
-    max_idle_interval: int
-
-    def __init__(self, connection_state_ttl: int, max_idle_interval: int):
-        self.connection_state_ttl = connection_state_ttl
-        self.max_idle_interval = max_idle_interval
-
-    @staticmethod
-    def from_dict(json_dict: dict):
-        return ConnectionDetails(json_dict.get('connectionStateTtl'), json_dict.get('maxIdleInterval'))
diff --git a/ably/util/helper.py b/ably/util/helper.py
index e221d1b..cead99d 100644
--- a/ably/util/helper.py
+++ b/ably/util/helper.py
@@ -2,8 +2,6 @@ import inspect
 import random
 import string
 import asyncio
-import time
-from typing import Callable
 
 
 def get_random_id():
@@ -15,24 +13,3 @@ def get_random_id():
 
 def is_callable_or_coroutine(value):
     return asyncio.iscoroutinefunction(value) or inspect.isfunction(value) or inspect.ismethod(value)
-
-
-def unix_time_ms():
-    return round(time.time_ns() / 1_000_000)
-
-
-class Timer:
-    def __init__(self, timeout: float, callback: Callable):
-        self._timeout = timeout
-        self._callback = callback
-        self._task = asyncio.create_task(self._job())
-
-    async def _job(self):
-        await asyncio.sleep(self._timeout / 1000)
-        if asyncio.iscoroutinefunction(self._callback):
-            await self._callback()
-        else:
-            self._callback()
-
-    def cancel(self):
-        self._task.cancel()


Test patch:
diff --git a/test/ably/eventemitter_test.py b/test/ably/eventemitter_test.py
index d981785..deda762 100644
--- a/test/ably/eventemitter_test.py
+++ b/test/ably/eventemitter_test.py
@@ -5,7 +5,7 @@ from test.ably.utils import BaseAsyncTestCase
 
 
 class TestEventEmitter(BaseAsyncTestCase):
-    async def asyncSetUp(self):
+    async def setUp(self):
         self.test_vars = await RestSetup.get_test_vars()
 
     async def test_connection_events(self):
diff --git a/test/ably/realtimechannel_test.py b/test/ably/realtimechannel_test.py
index 7881088..c95488c 100644
--- a/test/ably/realtimechannel_test.py
+++ b/test/ably/realtimechannel_test.py
@@ -9,7 +9,7 @@ from ably.util.exceptions import AblyException
 
 
 class TestRealtimeChannel(BaseAsyncTestCase):
-    async def asyncSetUp(self):
+    async def setUp(self):
         self.test_vars = await RestSetup.get_test_vars()
         self.valid_key_format = "api:key"
 
diff --git a/test/ably/realtimeconnection_test.py b/test/ably/realtimeconnection_test.py
index af7f0f2..f666a63 100644
--- a/test/ably/realtimeconnection_test.py
+++ b/test/ably/realtimeconnection_test.py
@@ -1,7 +1,7 @@
 import asyncio
 from ably.realtime.connection import ConnectionEvent, ConnectionState, ProtocolMessageAction
 import pytest
-from ably.util.exceptions import AblyException
+from ably.util.exceptions import AblyAuthException, AblyException
 from test.ably.restsetup import RestSetup
 from test.ably.utils import BaseAsyncTestCase
 from ably.transport.defaults import Defaults
@@ -38,7 +38,7 @@ class TestRealtimeConnection(BaseAsyncTestCase):
 
     async def test_auth_invalid_key(self):
         ably = await RestSetup.get_ably_realtime(key=self.valid_key_format)
-        with pytest.raises(AblyException) as exception:
+        with pytest.raises(AblyAuthException) as exception:
             await ably.connect()
         assert ably.connection.state == ConnectionState.FAILED
         assert ably.connection.error_reason == exception.value
@@ -62,7 +62,7 @@ class TestRealtimeConnection(BaseAsyncTestCase):
 
     async def test_connection_ping_failed(self):
         ably = await RestSetup.get_ably_realtime(key=self.valid_key_format)
-        with pytest.raises(AblyException) as exception:
+        with pytest.raises(AblyAuthException) as exception:
             await ably.connect()
         assert ably.connection.state == ConnectionState.FAILED
         assert ably.connection.error_reason == exception.value
@@ -115,7 +115,7 @@ class TestRealtimeConnection(BaseAsyncTestCase):
 
         ably.connection.on(ConnectionState.FAILED, on_state_change)
 
-        with pytest.raises(AblyException) as exception:
+        with pytest.raises(AblyAuthException) as exception:
             await ably.connect()
 
         assert len(failed_changes) == 1
@@ -288,47 +288,11 @@ class TestRealtimeConnection(BaseAsyncTestCase):
                 test_future.set_result(connection_state)
 
         ably.connection.on(ConnectionEvent.UPDATE, on_update)
-
-        async def on_transport_pending(transport):
-            await transport.on_protocol_message({'action': 4, "connectionDetails": {"connectionStateTtl": 200}})
-
-        ably.connection.connection_manager.on('transport.pending', on_transport_pending)
-
+        await ably.connection.connection_manager.on_protocol_message({'action': 4, "connectionDetails":
+                                                                      {"connectionStateTtl": 200}})
         state_change = await test_future
 
         assert state_change.previous == ConnectionState.CONNECTED
         assert state_change.current == ConnectionState.CONNECTED
         assert state_change.event == ConnectionEvent.UPDATE
         await ably.close()
-
-    async def test_max_idle_interval(self):
-        ably = await RestSetup.get_ably_realtime(realtime_request_timeout=2000)
-
-        test_future = asyncio.Future()
-
-        def on_transport_pending(transport):
-            original_on_protocol_message = transport.on_protocol_message
-
-            async def on_protocol_message(msg):
-                if msg["action"] == ProtocolMessageAction.CONNECTED:
-                    msg["connectionDetails"]["maxIdleInterval"] = 100
-
-                await original_on_protocol_message(msg)
-
-            transport.on_protocol_message = on_protocol_message
-
-        ably.connection.connection_manager.on('transport.pending', on_transport_pending)
-
-        def once_disconnected(state_change):
-            test_future.set_result(state_change)
-
-        ably.connection.once(ConnectionState.DISCONNECTED, once_disconnected)
-
-        state_change = await test_future
-
-        assert state_change.previous == ConnectionState.CONNECTED
-        assert state_change.current == ConnectionState.DISCONNECTED
-        assert state_change.reason.code == 80003
-        assert state_change.reason.status_code == 408
-
-        await ably.close()