From 299170c683502ae4a3b531e9578c1ff480b49927 Mon Sep 17 00:00:00 2001 From: Keith Date: Mon, 5 Oct 2026 18:37:48 -0400 Subject: [PATCH 1/2] Finish disconnecting immediately when disconnect() is called while reconnecting In the RECONNECTING state the previous socket has already closed, so close() never produces an onClose callback and the connection stayed in DISCONNECTING until the ping and pong timers ran out. Transition straight to DISCONNECTED instead, and cancel the pending reconnect timer. Fixes #376 --- .../websocket/WebSocketConnection.java | 14 ++++++-- .../websocket/WebSocketConnectionTest.java | 36 +++++++++++++++++++ 2 files changed, 48 insertions(+), 2 deletions(-) diff --git a/src/main/java/com/pusher/client/connection/websocket/WebSocketConnection.java b/src/main/java/com/pusher/client/connection/websocket/WebSocketConnection.java index 4a397d79..3be8998d 100644 --- a/src/main/java/com/pusher/client/connection/websocket/WebSocketConnection.java +++ b/src/main/java/com/pusher/client/connection/websocket/WebSocketConnection.java @@ -45,6 +45,7 @@ public class WebSocketConnection implements InternalConnection, WebSocketListene private final Consumer eventHandler; private String socketId; private int reconnectAttempts = 0; + private volatile Future reconnectTimer; public WebSocketConnection( final String url, @@ -93,7 +94,12 @@ private void tryConnecting() { @Override public void disconnect() { factory.queueOnEventThread(() -> { - if (canDisconnect()) { + if (state == ConnectionState.RECONNECTING) { + // The previous socket has already closed and won't call onClose, + // so finish disconnecting here instead of waiting for it. + updateState(ConnectionState.DISCONNECTING); + cancelTimeoutsAndTransitionToDisconnected(); + } else if (canDisconnect()) { updateState(ConnectionState.DISCONNECTING); underlyingConnection.close(); } @@ -253,7 +259,7 @@ private void tryReconnecting() { updateState(ConnectionState.RECONNECTING); long reconnectInterval = Math.min(maxReconnectionGap, reconnectAttempts * reconnectAttempts); - factory + reconnectTimer = factory .getTimers() .schedule( () -> { @@ -275,6 +281,10 @@ private boolean shouldReconnect(int code) { private void cancelTimeoutsAndTransitionToDisconnected() { activityTimer.cancelTimeouts(); + if (reconnectTimer != null) { + reconnectTimer.cancel(false); + reconnectTimer = null; + } factory.queueOnEventThread(() -> { if (state == ConnectionState.DISCONNECTING) { diff --git a/src/test/java/com/pusher/client/connection/websocket/WebSocketConnectionTest.java b/src/test/java/com/pusher/client/connection/websocket/WebSocketConnectionTest.java index afee2cd6..813227da 100644 --- a/src/test/java/com/pusher/client/connection/websocket/WebSocketConnectionTest.java +++ b/src/test/java/com/pusher/client/connection/websocket/WebSocketConnectionTest.java @@ -6,6 +6,7 @@ import static org.mockito.Mockito.any; import static org.mockito.Mockito.anyString; import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -34,6 +35,7 @@ import java.net.URI; import java.net.URISyntaxException; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; @@ -466,6 +468,40 @@ public Object answer(InvocationOnMock invocation) { assertEquals(ConnectionState.DISCONNECTED, connection.getState()); } + @Test + public void testDisconnectInReconnectingStateTransitionsToDisconnected() { + connect(); + connection.onClose(3999, "reason", true); + assertEquals(ConnectionState.RECONNECTING, connection.getState()); + + connection.disconnect(); + + assertEquals(ConnectionState.DISCONNECTED, connection.getState()); + verify(mockEventListener) + .onConnectionStateChange(new ConnectionStateChange(ConnectionState.RECONNECTING, ConnectionState.DISCONNECTING)); + verify(mockEventListener) + .onConnectionStateChange(new ConnectionStateChange(ConnectionState.DISCONNECTING, ConnectionState.DISCONNECTED)); + verify(mockUnderlyingConnection, never()).close(); + verify(factory).shutdownThreads(); + } + + @Test + public void testDisconnectInReconnectingStateCancelsPendingReconnect() { + final ScheduledFuture reconnectFuture = mock(ScheduledFuture.class); + when(factory.getTimers()).thenReturn(scheduledExecutorService); + doReturn(reconnectFuture) + .when(scheduledExecutorService) + .schedule(any(Runnable.class), any(Long.class), any(TimeUnit.class)); + + connection.connect(); + connection.onClose(500, "reason", true); + assertEquals(ConnectionState.RECONNECTING, connection.getState()); + + connection.disconnect(); + + verify(reconnectFuture).cancel(false); + } + /* end of tests */ private void connect() { From 21344a6d5c4f934aab550572652c5fb564afe78e Mon Sep 17 00:00:00 2001 From: Keith Date: Tue, 6 Oct 2026 14:36:18 -0400 Subject: [PATCH 2/2] Run the reconnect check on the event thread The reconnect task ran on the timers thread, so its RECONNECTING check could race with disconnect() and open a new socket after the connection was closed. Queue the check onto the event thread instead. Also remove the old socket's listener when disconnecting from RECONNECTING, and add a test that a reconnect task running after disconnect() does not open a new socket. --- .../websocket/WebSocketConnection.java | 6 ++++-- .../websocket/WebSocketConnectionTest.java | 18 ++++++++++++++++++ 2 files changed, 22 insertions(+), 2 deletions(-) diff --git a/src/main/java/com/pusher/client/connection/websocket/WebSocketConnection.java b/src/main/java/com/pusher/client/connection/websocket/WebSocketConnection.java index 3be8998d..81f650ce 100644 --- a/src/main/java/com/pusher/client/connection/websocket/WebSocketConnection.java +++ b/src/main/java/com/pusher/client/connection/websocket/WebSocketConnection.java @@ -97,6 +97,7 @@ public void disconnect() { if (state == ConnectionState.RECONNECTING) { // The previous socket has already closed and won't call onClose, // so finish disconnecting here instead of waiting for it. + underlyingConnection.removeWebSocketListener(); updateState(ConnectionState.DISCONNECTING); cancelTimeoutsAndTransitionToDisconnected(); } else if (canDisconnect()) { @@ -262,12 +263,13 @@ private void tryReconnecting() { reconnectTimer = factory .getTimers() .schedule( - () -> { + // Check the state on the event thread so it can't race with disconnect() + () -> factory.queueOnEventThread(() -> { if (state == ConnectionState.RECONNECTING) { underlyingConnection.removeWebSocketListener(); tryConnecting(); } - }, + }), reconnectInterval, TimeUnit.SECONDS ); diff --git a/src/test/java/com/pusher/client/connection/websocket/WebSocketConnectionTest.java b/src/test/java/com/pusher/client/connection/websocket/WebSocketConnectionTest.java index 813227da..b8106f9e 100644 --- a/src/test/java/com/pusher/client/connection/websocket/WebSocketConnectionTest.java +++ b/src/test/java/com/pusher/client/connection/websocket/WebSocketConnectionTest.java @@ -25,6 +25,7 @@ import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.invocation.InvocationOnMock; import org.mockito.runners.MockitoJUnitRunner; @@ -502,6 +503,23 @@ public void testDisconnectInReconnectingStateCancelsPendingReconnect() { verify(reconnectFuture).cancel(false); } + @Test + public void testReconnectThatRunsAfterDisconnectDoesNotReconnect() throws SSLException { + final ArgumentCaptor reconnect = ArgumentCaptor.forClass(Runnable.class); + when(factory.getTimers()).thenReturn(scheduledExecutorService); + + connection.connect(); + connection.onClose(500, "reason", true); + verify(scheduledExecutorService).schedule(reconnect.capture(), any(Long.class), any(TimeUnit.class)); + + connection.disconnect(); + // Simulate the reconnect task already running when disconnect() cancelled it + reconnect.getValue().run(); + + verify(factory, times(1)).newWebSocketClientWrapper(any(URI.class), any(Proxy.class), any(WebSocketConnection.class)); + assertEquals(ConnectionState.DISCONNECTED, connection.getState()); + } + /* end of tests */ private void connect() {