fix: ensure read_loop properly closed when disposing transport

Author: owenpearson

ably/transport/websockettransport.py

@@ -229,18 +229,38 @@ def on_read_loop_done(self, task: asyncio.Task):
 
     async def dispose(self):
         self.is_disposed = True
+
+        # Cancel tasks but don't await them yet to avoid deadlock
+        tasks_to_await = []
+
         if self.read_loop:
             self.read_loop.cancel()
+            tasks_to_await.append(self.read_loop)
         if self.ws_connect_task:
             self.ws_connect_task.cancel()
+            tasks_to_await.append(self.ws_connect_task)
         if self.idle_timer:
             self.idle_timer.cancel()
+
+        # Schedule cleanup of cancelled tasks in the background to avoid blocking dispose()
+        # This prevents deadlock when dispose() is called from within these tasks
+        if tasks_to_await:
+            asyncio.create_task(self._cleanup_tasks(tasks_to_await))
+
         if self.websocket:
             try:
                 await self.websocket.close()
             except asyncio.CancelledError:
                 return
 
+    async def _cleanup_tasks(self, tasks):
+        """Wait for cancelled tasks to complete their cleanup."""
+        for task in tasks:
+            try:
+                await task
+            except Exception:
+                pass  # Ignore all exceptions from cancelled/failed tasks
+
     async def close(self):
         await self.send({'action': ProtocolMessageAction.CLOSE})