Repository: ably/ably-python (python)

Issue:
Propagate connection interruptions to realtime channels
If a `RealtimeChannel` is in an active state (ie `ATTACHING`, `ATTACHED`, `DETACHING`, or `SUSPENDED`) the channel state should change upon the following connection states:
- If the connection becomes `CLOSING` the channel should become `DETACHED`
- If the connection becomes `CLOSED` the channel should become `DETACHED`
- If the connection becomes `FAILED` the channel should become `FAILED`
- If the connection becomes `SUSPENDED` the channel should become `SUSPENDED`

Pull request description:
Propagate connection interruption to realtime channels
Resolves #417 

Patch:
diff --git a/ably/http/http.py b/ably/http/http.py
index d53b540..e2607ca 100644
--- a/ably/http/http.py
+++ b/ably/http/http.py
@@ -145,7 +145,7 @@ class Http:
 
     def get_rest_hosts(self):
         hosts = self.options.get_rest_hosts()
-        host = self.__host or self.options.fallback_realtime_host
+        host = self.__host
         if host is None:
             return hosts
 
diff --git a/ably/realtime/connection.py b/ably/realtime/connection.py
index db49f98..69a7aa5 100644
--- a/ably/realtime/connection.py
+++ b/ably/realtime/connection.py
@@ -166,7 +166,6 @@ class ConnectionManager(EventEmitter):
         self.retry_timer: Timer | None = None
         self.connect_base_task: asyncio.Task | None = None
         self.disconnect_transport_task: asyncio.Task | None = None
-        self.__fallback_hosts = self.options.get_fallback_realtime_hosts()
         super().__init__()
 
     def enact_state_change(self, state, reason=None):
@@ -241,21 +240,6 @@ class ConnectionManager(EventEmitter):
         else:
             self.notify_state(ConnectionState.CONNECTED)
 
-    def on_disconnected(self, msg: dict):
-        error = msg.get("error")
-        exception = AblyException(error.get('message'), error.get('statusCode'), error.get('code'))
-        self.notify_state(ConnectionState.DISCONNECTED, exception)
-        if error:
-            error_status_code = error.get("statusCode")
-            if error_status_code >= 500 or error_status_code <= 504:  # RTN17f1
-                if len(self.__fallback_hosts) > 0:
-                    res = asyncio.create_task(self.connect_with_fallback_hosts(self.__fallback_hosts))
-                    if not res:
-                        return
-                    self.notify_state(self.__fail_state, reason=res)
-                else:
-                    log.info("No fallback host to try for disconnected protocol message")
-
     async def on_error(self, msg: dict, exception: AblyException):
         if msg.get('channel') is None:  # RTN15i
             self.enact_state_change(ConnectionState.FAILED, exception)
@@ -310,42 +294,8 @@ class ConnectionManager(EventEmitter):
         self.start_transition_timer(ConnectionState.CONNECTING)
         self.connect_base_task = asyncio.create_task(self.connect_base())
 
-    async def connect_with_fallback_hosts(self, fallback_hosts: list):
-        for host in fallback_hosts:
-            try:
-                if self.check_connection():
-                    await self.try_host(host)
-                    return
-                else:
-                    message = "Unable to connect, network unreachable"
-                    log.exception(message)
-                    exception = AblyException(message, status_code=404, code=80003)
-                    self.notify_state(self.__fail_state, exception)
-                    return
-            except Exception as exc:
-                exception = exc
-                log.exception(f'Connection to {host} failed, reason={exception}')
-        log.exception("No more fallback hosts to try")
-        return exception
-
     async def connect_base(self):
-        fallback_hosts = self.__fallback_hosts
-        primary_host = self.options.get_realtime_host()
-        try:
-            await self.try_host(primary_host)
-            return
-        except Exception as exception:
-            log.exception(f'Connection to {primary_host} failed, reason={exception}')
-            if len(fallback_hosts) > 0:
-                log.info("Attempting connection to fallback host(s)")
-                resp = await self.connect_with_fallback_hosts(fallback_hosts)
-                if not resp:
-                    return
-                exception = resp
-            self.notify_state(self.__fail_state, reason=exception)
-
-    async def try_host(self, host):
-        self.transport = WebSocketTransport(self, host)
+        self.transport = WebSocketTransport(self)
         self._emit('transport.pending', self.transport)
         self.transport.connect()
 
@@ -366,20 +316,14 @@ class ConnectionManager(EventEmitter):
 
         self.transport.once('connected', on_transport_connected)
         self.transport.once('failed', on_transport_failed)
-        #  Fix asyncio CancelledError in python 3.7
+
         try:
             await future
-        except asyncio.CancelledError:
-            return
+        except Exception as exception:
+            self.notify_state(self.__fail_state, reason=exception)
 
     def notify_state(self, state: ConnectionState, reason=None):
-        # RTN15a
-        retry_immediately = state == ConnectionState.DISCONNECTED and self.__state == ConnectionState.CONNECTED
-
-        log.info(
-            f'ConnectionManager.notify_state(): new state: {state}'
-            + ('; will retry immediately' if retry_immediately else '')
-        )
+        log.info(f'ConnectionManager.notify_state(): new state: {state}')
 
         if state == self.__state:
             return
@@ -387,18 +331,24 @@ class ConnectionManager(EventEmitter):
         self.cancel_transition_timer()
         self.check_suspend_timer(state)
 
-        if retry_immediately:
-            self.options.loop.call_soon(self.request_state, ConnectionState.CONNECTING)
-        elif state == ConnectionState.DISCONNECTED:
+        if state == ConnectionState.DISCONNECTED:
             self.start_retry_timer(self.options.disconnected_retry_timeout)
         elif state == ConnectionState.SUSPENDED:
             self.start_retry_timer(self.options.suspended_retry_timeout)
 
-        if (state == ConnectionState.DISCONNECTED and not retry_immediately) or state == ConnectionState.SUSPENDED:
+        if state == ConnectionState.DISCONNECTED or state == ConnectionState.SUSPENDED:
             self.disconnect_transport()
 
         self.enact_state_change(state, reason)
 
+        if state in (
+            ConnectionState.CLOSING,
+            ConnectionState.CLOSED,
+            ConnectionState.SUSPENDED,
+            ConnectionState.FAILED,
+        ):
+            self.ably.channels._propagate_connection_interruption(state, reason)
+
     def start_transition_timer(self, state: ConnectionState, fail_state=None):
         log.debug(f'ConnectionManager.start_transition_timer(): transition state = {state}')
 
diff --git a/ably/realtime/realtime.py b/ably/realtime/realtime.py
index d1499b1..6fd0bc8 100644
--- a/ably/realtime/realtime.py
+++ b/ably/realtime/realtime.py
@@ -4,7 +4,7 @@ from ably.realtime.connection import Connection, ConnectionState
 from ably.rest.auth import Auth
 from ably.rest.rest import AblyRest
 from ably.types.options import Options
-from ably.realtime.realtime_channel import RealtimeChannel
+from ably.realtime.realtime_channel import ChannelState, RealtimeChannel
 
 
 log = logging.getLogger(__name__)
@@ -218,3 +218,23 @@ class Channels:
             return
 
         channel._on_message(msg)
+
+    def _propagate_connection_interruption(self, state: ConnectionState, reason):
+        from_channel_states = (
+            ChannelState.ATTACHING,
+            ChannelState.ATTACHED,
+            ChannelState.DETACHING,
+            ChannelState.SUSPENDED,
+        )
+
+        connection_to_channel_state = {
+            ConnectionState.CLOSING: ChannelState.DETACHED,
+            ConnectionState.CLOSED: ChannelState.DETACHED,
+            ConnectionState.FAILED: ChannelState.FAILED,
+            ConnectionState.SUSPENDED: ChannelState.SUSPENDED,
+        }
+
+        for name in self.all.keys():
+            channel = self.all[name]
+            if channel.state in from_channel_states:
+                channel._notify_state(connection_to_channel_state[state], reason)
diff --git a/ably/realtime/realtime_channel.py b/ably/realtime/realtime_channel.py
index f832a80..10b88c5 100644
--- a/ably/realtime/realtime_channel.py
+++ b/ably/realtime/realtime_channel.py
@@ -58,12 +58,9 @@ class RealtimeChannel(EventEmitter, Channel):
     def __init__(self, realtime, name):
         EventEmitter.__init__(self)
         self.__name = name
-        self.__attach_future = None
-        self.__detach_future = None
         self.__realtime = realtime
         self.__state = ChannelState.INITIALIZED
         self.__message_emitter = EventEmitter()
-        self.__timeout_in_secs = self.__realtime.options.realtime_request_timeout / 1000
         self.__state_timer: Timer | None = None
 
         # Used to listen to state changes internally, if we use the public event emitter interface then internals
diff --git a/ably/realtime/websockettransport.py b/ably/realtime/websockettransport.py
index fef9304..e90a57d 100644
--- a/ably/realtime/websockettransport.py
+++ b/ably/realtime/websockettransport.py
@@ -24,7 +24,6 @@ log = logging.getLogger(__name__)
 class ProtocolMessageAction(IntEnum):
     HEARTBEAT = 0
     CONNECTED = 4
-    DISCONNECTED = 6
     CLOSE = 7
     CLOSED = 8
     ERROR = 9
@@ -36,7 +35,7 @@ class ProtocolMessageAction(IntEnum):
 
 
 class WebSocketTransport(EventEmitter):
-    def __init__(self, connection_manager: ConnectionManager, host: str):
+    def __init__(self, connection_manager: ConnectionManager):
         self.websocket: WebSocketClientProtocol | None = None
         self.read_loop: asyncio.Task | None = None
         self.connect_task: asyncio.Task | None = None
@@ -48,7 +47,6 @@ class WebSocketTransport(EventEmitter):
         self.last_activity = None
         self.max_idle_interval = None
         self.is_disposed = False
-        self.host = host
         super().__init__()
 
     def connect(self):
@@ -56,7 +54,7 @@ class WebSocketTransport(EventEmitter):
         protocol_version = Defaults.protocol_version
         params = {"key": self.connection_manager.ably.key, "v": protocol_version}
         query_params = urllib.parse.urlencode(params)
-        ws_url = (f'wss://{self.host}?{query_params}')
+        ws_url = (f'wss://{self.connection_manager.options.get_realtime_host()}?{query_params}')
         log.info(f'connect(): attempting to connect to {ws_url}')
         self.ws_connect_task = asyncio.create_task(self.ws_connect(ws_url, headers))
         self.ws_connect_task.add_done_callback(self.on_ws_connect_done)
@@ -103,11 +101,7 @@ class WebSocketTransport(EventEmitter):
                 self.max_idle_interval = max_idle_interval + self.options.realtime_request_timeout
                 self.on_activity()
             self.is_connected = True
-            if self.host != self.options.get_realtime_host():  # RTN17e
-                self.options.fallback_realtime_host = self.host
             self.connection_manager.on_connected(connection_details)
-        elif action == ProtocolMessageAction.DISCONNECTED:
-            self.connection_manager.on_disconnected(msg)
         elif action == ProtocolMessageAction.CLOSED:
             if self.ws_connect_task:
                 self.ws_connect_task.cancel()
diff --git a/ably/types/options.py b/ably/types/options.py
index 750b91a..90d112c 100644
--- a/ably/types/options.py
+++ b/ably/types/options.py
@@ -46,9 +46,6 @@ class Options(AuthOptions):
             from ably import api_version
             idempotent_rest_publishing = api_version >= '1.2'
 
-        if environment is None:
-            environment = Defaults.environment
-
         self.__client_id = client_id
         self.__log_level = log_level
         self.__tls = tls
@@ -75,7 +72,6 @@ class Options(AuthOptions):
         self.__connection_state_ttl = connection_state_ttl
         self.__suspended_retry_timeout = suspended_retry_timeout
         self.__connectivity_check_url = connectivity_check_url
-        self.__fallback_realtime_host = None
 
         self.__rest_hosts = self.__get_rest_hosts()
         self.__realtime_hosts = self.__get_realtime_hosts()
@@ -246,14 +242,6 @@ class Options(AuthOptions):
     def connectivity_check_url(self):
         return self.__connectivity_check_url
 
-    @property
-    def fallback_realtime_host(self):
-        return self.__fallback_realtime_host
-
-    @fallback_realtime_host.setter
-    def fallback_realtime_host(self, value):
-        self.__fallback_realtime_host = value
-
     def __get_rest_hosts(self):
         """
         Return the list of hosts as they should be tried. First comes the main
@@ -266,6 +254,8 @@ class Options(AuthOptions):
             host = Defaults.rest_host
 
         environment = self.environment
+        if environment is None:
+            environment = Defaults.environment
 
         http_max_retry_count = self.http_max_retry_count
         if http_max_retry_count is None:
@@ -302,7 +292,6 @@ class Options(AuthOptions):
         # Shuffle
         fallback_hosts = list(fallback_hosts)
         random.shuffle(fallback_hosts)
-        self.__fallback_hosts = fallback_hosts
 
         # First main host
         hosts = [host] + fallback_hosts
@@ -311,13 +300,11 @@ class Options(AuthOptions):
 
     def __get_realtime_hosts(self):
         if self.realtime_host is not None:
-            host = self.realtime_host
-        elif self.environment != "production":
-            host = f'{self.environment}-{Defaults.realtime_host}'
+            return self.realtime_host
+        elif self.environment is not None:
+            return f'{self.environment}-{Defaults.realtime_host}'
         else:
-            host = Defaults.realtime_host
-
-        return [host] + self.__fallback_hosts
+            return Defaults.realtime_host
 
     def get_rest_hosts(self):
         return self.__rest_hosts
@@ -325,14 +312,8 @@ class Options(AuthOptions):
     def get_rest_host(self):
         return self.__rest_hosts[0]
 
-    def get_realtime_hosts(self):
-        return self.__realtime_hosts
-
     def get_realtime_host(self):
-        return self.__realtime_hosts[0]
+        return self.__realtime_hosts
 
     def get_fallback_rest_hosts(self):
         return self.__rest_hosts[1:]
-
-    def get_fallback_realtime_hosts(self):
-        return self.__realtime_hosts[1:]


Test patch:
diff --git a/test/ably/realtimechannel_test.py b/test/ably/realtimechannel_test.py
index e171e87..cbe16d4 100644
--- a/test/ably/realtimechannel_test.py
+++ b/test/ably/realtimechannel_test.py
@@ -3,7 +3,7 @@ import pytest
 from ably.realtime.realtime_channel import ChannelState
 from ably.types.message import Message
 from test.ably.restsetup import RestSetup
-from test.ably.utils import BaseAsyncTestCase
+from test.ably.utils import BaseAsyncTestCase, random_string
 from ably.realtime.connection import ConnectionState, ProtocolMessageAction
 from ably.util.exceptions import AblyException
 
@@ -252,3 +252,34 @@ class TestRealtimeChannel(BaseAsyncTestCase):
         assert exception.value.code == 90007
         assert exception.value.status_code == 408
         await ably.close()
+
+    async def test_channel_detached_once_connection_closed(self):
+        ably = await RestSetup.get_ably_realtime(realtime_request_timeout=2000)
+        await ably.connection.once_async(ConnectionState.CONNECTED)
+        channel = ably.channels.get(random_string(5))
+        await channel.attach()
+
+        await ably.close()
+        assert channel.state == ChannelState.DETACHED
+
+    async def test_channel_failed_once_connection_failed(self):
+        ably = await RestSetup.get_ably_realtime(realtime_request_timeout=2000)
+        await ably.connection.once_async(ConnectionState.CONNECTED)
+        channel = ably.channels.get(random_string(5))
+        await channel.attach()
+
+        ably.connection.connection_manager.notify_state(ConnectionState.SUSPENDED)
+        assert channel.state == ChannelState.SUSPENDED
+
+        await ably.close()
+
+    async def test_channel_suspended_once_connection_suspended(self):
+        ably = await RestSetup.get_ably_realtime(realtime_request_timeout=2000)
+        await ably.connection.once_async(ConnectionState.CONNECTED)
+        channel = ably.channels.get(random_string(5))
+        await channel.attach()
+
+        ably.connection.connection_manager.notify_state(ConnectionState.FAILED)
+        assert channel.state == ChannelState.FAILED
+
+        await ably.close()
diff --git a/test/ably/realtimeconnection_test.py b/test/ably/realtimeconnection_test.py
index 8f13f31..916c119 100644
--- a/test/ably/realtimeconnection_test.py
+++ b/test/ably/realtimeconnection_test.py
@@ -264,61 +264,3 @@ class TestRealtimeConnection(BaseAsyncTestCase):
         assert state_change.reason.status_code == 408
 
         await ably.close()
-
-    # RTN15a
-    async def test_retry_immediately_upon_unexpected_disconnection(self):
-        # Set timeouts to 500s so that if the client uses retry delay the test will fail with a timeout
-        ably = await RestSetup.get_ably_realtime(
-            disconnected_retry_timeout=500_000,
-            suspended_retry_timeout=500_000
-        )
-
-        # Wait for the client to connect
-        await ably.connection.once_async(ConnectionState.CONNECTED)
-
-        # Simulate random loss of connection
-        assert ably.connection.connection_manager.transport
-        await ably.connection.connection_manager.transport.dispose()
-        ably.connection.connection_manager.notify_state(ConnectionState.DISCONNECTED)
-        assert ably.connection.state == ConnectionState.DISCONNECTED
-
-        # Wait for the client to connect again
-        await ably.connection.once_async(ConnectionState.CONNECTED)
-        await ably.close()
-
-    async def test_fallback_host(self):
-        fallback_host = 'sandbox-realtime.ably.io'
-        ably = await RestSetup.get_ably_realtime(realtime_host="iamnotahost", disconnected_retry_timeout=400000,
-                                                 fallback_hosts=[fallback_host])
-        await ably.connection.once_async(ConnectionState.CONNECTED)
-        assert ably.connection.connection_manager.transport.host == fallback_host
-        assert ably.options.fallback_realtime_host == fallback_host
-        await ably.close()
-
-    async def test_fallback_host_no_connection(self):
-        fallback_host = 'sandbox-realtime.ably.io'
-        ably = await RestSetup.get_ably_realtime(realtime_host="iamnotahost", disconnected_retry_timeout=400000,
-                                                 fallback_hosts=[fallback_host])
-
-        def check_connection():
-            return False
-
-        ably.connection.connection_manager.check_connection = check_connection
-
-        await ably.connection.once_async(ConnectionState.DISCONNECTED)
-        assert ably.connection.connection_manager.transport.host == "iamnotahost"
-        await ably.close()
-
-    async def test_fallback_host_disconnected_protocol_msg(self):
-        fallback_host = 'sandbox-realtime.ably.io'
-        ably = await RestSetup.get_ably_realtime(realtime_host="iamnotahost", disconnected_retry_timeout=400000,
-                                                 fallback_hosts=[fallback_host])
-
-        async def on_transport_pending(transport):
-            await transport.on_protocol_message({'action': 6, "error": {"statusCode": 500, "code": 500}})
-
-        ably.connection.connection_manager.once('transport.pending', on_transport_pending)
-
-        await ably.connection.once_async(ConnectionState.CONNECTED)
-        assert ably.connection.connection_manager.transport.host == fallback_host
-        await ably.close()