test: add unit tests for WebSocketTransport lifecycle and robustness

- Test edge cases like multiple `connect()` calls, forced closures, and redundant `onClose` events.

Author: ttypic

lib/src/main/java/io/ably/lib/transport/ConnectionManager.java

@@ -31,6 +31,7 @@
 import io.ably.lib.types.ClientOptions;
 import io.ably.lib.types.ConnectionDetails;
 import io.ably.lib.types.ErrorInfo;
+import io.ably.lib.types.Param;
 import io.ably.lib.types.ProtocolMessage;
 import io.ably.lib.types.ProtocolSerializer;
 import io.ably.lib.util.Log;
@@ -857,6 +858,13 @@ public void requestState(StateIndication state) {
         requestState(null, state);
     }
 
+    /**
+     * Get query params representing the current authentication method and credentials.
+     */
+    Param[] getAuthParams() throws AblyException {
+        return ably.auth.getAuthParams();
+    }
+
     /**
      * Determines if the given WebSocketTransport instance is the currently active transport.
      *

lib/src/main/java/io/ably/lib/transport/WebSocketTransport.java

@@ -109,7 +109,7 @@ public void connect(ConnectListener connectListener) {
             boolean isTls = params.options.tls;
             String wsScheme = isTls ? "wss://" : "ws://";
             wsUri = wsScheme + params.host + ':' + params.port + "/";
-            Param[] authParams = connectionManager.ably.auth.getAuthParams();
+            Param[] authParams = connectionManager.getAuthParams();
             Param[] connectParams = params.getConnectParams(authParams);
             if (connectParams.length > 0)
                 wsUri = HttpUtils.encodeParams(wsUri, connectParams);
@@ -415,7 +415,7 @@ private void schedule(TimerTask task, long delay) {
             try {
                 timer.schedule(task, delay);
             } catch (IllegalStateException ise) {
-                Log.w(TAG, "Timer has already been canceled", ise);
+                Log.w(TAG, "Timer has already has been canceled", ise);
             }
         }
 
@@ -439,7 +439,7 @@ private void onActivityTimerExpiry() {
         }
 
         private long getActivityTimeout() {
-            return connectionManager.maxIdleInterval + connectionManager.ably.options.realtimeRequestTimeout;
+            return connectionManager.maxIdleInterval + params.options.realtimeRequestTimeout;
         }
     }
 

lib/src/test/java/io/ably/lib/test/common/Helpers.java

@@ -969,6 +969,16 @@ public static boolean equalNullableStrings(String one, String two) {
         return (one == null) ? (two == null) : one.equals(two);
     }
 
+    public static void setPrivateField(Object object, String fieldName, Object value) {
+        try {
+            Field connectionStateField = object.getClass().getDeclaredField(fieldName);
+            connectionStateField.setAccessible(true);
+            connectionStateField.set(object, value);
+        } catch (Exception e) {
+            fail("Failed accessing " + fieldName + " with error " + e);
+        }
+    }
+
     public static class RawHttpRequest {
         public String id;
         public URL url;

lib/src/test/java/io/ably/lib/transport/WebSocketTransportTest.java

@@ -0,0 +1,158 @@
+package io.ably.lib.transport;
+
+import io.ably.lib.network.WebSocketClient;
+import io.ably.lib.network.WebSocketEngine;
+import io.ably.lib.network.WebSocketListener;
+import io.ably.lib.realtime.AblyRealtime;
+import io.ably.lib.realtime.CompletionListener;
+import io.ably.lib.test.common.Helpers;
+import io.ably.lib.test.common.ParameterizedTest;
+import io.ably.lib.test.util.EmptyPlatformAgentProvider;
+import io.ably.lib.transport.ITransport.TransportParams;
+import io.ably.lib.types.AblyException;
+import io.ably.lib.types.ClientOptions;
+import io.ably.lib.types.Param;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.Timeout;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
+
+/**
+ * Unit tests for WebSocketTransport, specifically testing activity timer behavior
+ * when WebSocket close operations get stuck or fail to trigger onClose handlers.
+ */
+public class WebSocketTransportTest {
+
+    private ConnectionManager mockConnectionManager;
+
+    private WebSocketEngine mockEngine;
+
+    private WebSocketTransport transport;
+
+    private WebSocketClient mockWebSocketClient;
+
+    private TransportParams transportParams;
+
+    @Before
+    public void setUp() throws Exception {
+        mockConnectionManager = mock(ConnectionManager.class);
+        mockEngine = mock(WebSocketEngine.class);
+        mockWebSocketClient = mock(WebSocketClient.class);
+        when(mockEngine.isPingListenerSupported()).thenReturn(true);
+        when(mockEngine.create(any(), any())).thenReturn(mockWebSocketClient);
+        when(mockConnectionManager.getAuthParams()).thenReturn(new Param[] {});
+
+        mockConnectionManager.maxIdleInterval = 10;
+
+        // Setup transport params
+        transportParams = new TransportParams(new ClientOptions(), new EmptyPlatformAgentProvider());
+        transportParams.host = "realtime.ably.io";
+        transportParams.port = 443;
+        transportParams.options.realtimeRequestTimeout = 10;
+    }
+
+    private WebSocketTransport createWebSocketTransport() {
+        WebSocketTransport transport = new WebSocketTransport(transportParams, mockConnectionManager);
+        Helpers.setPrivateField(transport, "webSocketEngine", mockEngine);
+        return transport;
+    }
+
+    @Test
+    public void throwExceptionsIfConnectCalledTwice() {
+        final WebSocketTransport transport = createWebSocketTransport();
+        ITransport.ConnectListener connectListener = mock(ITransport.ConnectListener.class);
+        transport.connect(connectListener);
+        assertThrows(IllegalStateException.class, () ->
+            transport.connect(connectListener)
+        );
+    }
+
+    @Test
+    public void shouldCallCancelIfNotClosedGracefully() {
+        AtomicReference<WebSocketListener> webSocketListenerRef = new AtomicReference<>();
+
+        when(mockEngine.create(any(), any())).thenAnswer(invocation -> {
+            webSocketListenerRef.set(invocation.getArgumentAt(1, WebSocketListener.class));
+            return mockWebSocketClient;
+        });
+
+        doAnswer(invocation -> {
+            webSocketListenerRef.get().onClose(
+                invocation.getArgumentAt(0, Integer.class),
+                invocation.getArgumentAt(1, String.class)
+            );
+            return null;
+        }).when(mockWebSocketClient).cancel(anyInt(), anyString());
+
+        final WebSocketTransport transport = createWebSocketTransport();
+        ITransport.ConnectListener connectListener = mock(ITransport.ConnectListener.class);
+        transport.connect(connectListener);
+        transport.close();
+        // check that we tried to close gracefully
+        verify(mockWebSocketClient).close();
+        // check that we closed forcibly at the end
+        verify(mockWebSocketClient, timeout(1_000)).cancel(eq(1006), anyString());
+        // verify that we call listener at the end
+        verify(connectListener).onTransportUnavailable(eq(transport), any());
+    }
+
+    /**
+     * `onClose` can be called twice, e.g. from activity timer force close and from manual `close()`
+     * It shouldn't result in any exceptions
+     */
+    @Test
+    public void shouldNotThrowExceptionIfSeveralCloseEventsHappened() {
+        AtomicReference<WebSocketListener> listenerRef = new AtomicReference<>();
+
+        when(mockEngine.create(any(), any())).thenAnswer(invocation -> {
+            listenerRef.set(invocation.getArgumentAt(1, WebSocketListener.class));
+            return mockWebSocketClient;
+        });
+
+
+        final WebSocketTransport transport = createWebSocketTransport();
+        ITransport.ConnectListener connectListener = mock(ITransport.ConnectListener.class);
+        transport.connect(connectListener);
+
+        listenerRef.get().onClose(1000, "OK");
+        listenerRef.get().onClose(1006, "Abnormal close");
+
+        verify(connectListener, times(2)).onTransportUnavailable(eq(transport), any());
+    }
+
+    /**
+     * Calling `close()` on transport triggers the activity timer.
+     * Test checks that if it has been disposed it won't do anything.
+     */
+    @Test
+    public void shouldNotThrowExceptionIfCloseCalledOnAlreadyClosedTransport() {
+        AtomicReference<WebSocketListener> listenerRef = new AtomicReference<>();
+
+        when(mockEngine.create(any(), any())).thenAnswer(invocation -> {
+            listenerRef.set(invocation.getArgumentAt(1, WebSocketListener.class));
+            return mockWebSocketClient;
+        });
+
+
+        final WebSocketTransport transport = createWebSocketTransport();
+        ITransport.ConnectListener connectListener = mock(ITransport.ConnectListener.class);
+        transport.connect(connectListener);
+
+        listenerRef.get().onClose(1006, "Abnormal close");
+        transport.close();
+        verify(connectListener, timeout(1_000)).onTransportUnavailable(eq(transport), any());
+    }
+}