From 9a4bc5c355b73dd391f411d0270df92f8d097f66 Mon Sep 17 00:00:00 2001 From: elliotoakley <328120662+elliotoakley@users.noreply.github.com> Date: Sun, 13 Sep 2026 14:06:22 +0500 Subject: [PATCH] fix(initiator): reset socket address index only after successful FIX logon (#1260) --- .../mina/initiator/IoSessionInitiator.java | 24 ++- .../initiator/IoSessionInitiatorTest.java | 189 ++++++++++++++++++ 2 files changed, 210 insertions(+), 3 deletions(-) diff --git a/quickfixj-core/src/main/java/quickfix/mina/initiator/IoSessionInitiator.java b/quickfixj-core/src/main/java/quickfix/mina/initiator/IoSessionInitiator.java index 84adf56e1..95a6d0357 100644 --- a/quickfixj-core/src/main/java/quickfix/mina/initiator/IoSessionInitiator.java +++ b/quickfixj-core/src/main/java/quickfix/mina/initiator/IoSessionInitiator.java @@ -44,6 +44,7 @@ import quickfix.mina.ssl.SSLConfig; import quickfix.mina.ssl.SSLContextFactory; import quickfix.mina.ssl.SSLSupport; +import quickfix.SessionStateListener; import javax.net.ssl.SSLContext; import java.io.IOException; @@ -61,6 +62,7 @@ public class IoSessionInitiator { private final static long CONNECT_POLL_TIMEOUT = 2000L; private final ScheduledExecutorService executor; private final ConnectTask reconnectTask; + private final SessionStateListener stateListener; private final Logger log = LoggerFactory.getLogger(getClass()); private Future reconnectFuture; @@ -90,6 +92,13 @@ public class IoSessionInitiator { throw new ConfigError(e); } + stateListener = new SessionStateListener() { + @Override + public void onLogon(SessionID sessionID) { + reconnectTask.resetSocketAddressIndex(); + } + }; + fixSession.getLog().onEvent("Configured socket addresses for session: " + Arrays.asList(socketAddresses)); } @@ -113,6 +122,7 @@ private static class ConnectTask implements Runnable { private long lastReconnectAttemptTime; private long lastConnectTime; private int nextSocketAddressIndex; + private volatile boolean socketAddressIndexResetRequested; private int connectionFailureCount; private ConnectFuture connectFuture; @@ -225,6 +235,10 @@ public String getSniHostName(SSLConfig sslConfig) { @Override public void run() { + if (socketAddressIndexResetRequested) { + nextSocketAddressIndex = 0; + socketAddressIndexResetRequested = false; + } resetIoConnector(); try { if (connectFuture == null) { @@ -261,7 +275,6 @@ private void pollConnectFuture() { if (connectFuture.getSession() != null) { ioSession = connectFuture.getSession(); connectionFailureCount = 0; - nextSocketAddressIndex = 0; lastConnectTime = System.currentTimeMillis(); connectFuture = null; } else { @@ -369,6 +382,10 @@ private void resetIoConnector() { } } } + + private void resetSocketAddressIndex() { + socketAddressIndexResetRequested = true; + } } synchronized void start() { @@ -376,8 +393,8 @@ synchronized void start() { // The following logon reenabled the session. The actual logon will take // place as a side-effect of the session timer task (not the reconnect task). reconnectTask.getFixSession().logon(); // only enables the session - reconnectFuture = executor - .scheduleWithFixedDelay(reconnectTask, 0, 1, TimeUnit.SECONDS); + reconnectTask.getFixSession().addStateListener(stateListener); + reconnectFuture = executor.scheduleWithFixedDelay(reconnectTask, 0, 1, TimeUnit.SECONDS); } } @@ -385,6 +402,7 @@ synchronized void stop() { if (reconnectFuture != null) { reconnectFuture.cancel(true); reconnectFuture = null; + reconnectTask.getFixSession().removeStateListener(stateListener); } SessionConnector.closeManagedSessionsAndDispose(reconnectTask.ioConnector, true, log); } diff --git a/quickfixj-core/src/test/java/quickfix/mina/initiator/IoSessionInitiatorTest.java b/quickfixj-core/src/test/java/quickfix/mina/initiator/IoSessionInitiatorTest.java index 20f9bb86e..8079901c0 100644 --- a/quickfixj-core/src/test/java/quickfix/mina/initiator/IoSessionInitiatorTest.java +++ b/quickfixj-core/src/test/java/quickfix/mina/initiator/IoSessionInitiatorTest.java @@ -5,6 +5,7 @@ import quickfix.Log; import quickfix.Session; import quickfix.SessionSettings; +import quickfix.SessionStateListener; import quickfix.mina.EventHandlingStrategy; import quickfix.mina.HostResolutionStrategy; import quickfix.mina.NetworkingOptions; @@ -17,6 +18,16 @@ import static org.junit.Assert.assertEquals; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; +import static org.mockito.Mockito.verify; +import org.mockito.ArgumentCaptor; + +import java.net.ServerSocket; +import java.net.Socket; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.Executors; + +import static org.junit.Assert.assertTrue; public class IoSessionInitiatorTest { @@ -35,6 +46,118 @@ public void shouldNotConfigureProxyForVmPipeTransport() throws Exception { assertProxyIsNotConfigured(new VmPipeAddress(5001), "invalid", 8080); } + @Test + public void shouldTryNextSocketAddressWhenDisconnectedBeforeLogon() throws Exception { + try (ServerSocket firstServer = new ServerSocket(0); + ServerSocket secondServer = new ServerSocket(0)) { + + CountDownLatch firstConnection = new CountDownLatch(1); + CountDownLatch secondConnection = new CountDownLatch(1); + + Thread firstServerThread = new Thread(() -> { + try (Socket socket = firstServer.accept()) { + firstConnection.countDown(); + } catch (Exception ignored) { + } + }); + + Thread secondServerThread = new Thread(() -> { + try (Socket socket = secondServer.accept()) { + secondConnection.countDown(); + } catch (Exception ignored) { + } + }); + + firstServerThread.start(); + secondServerThread.start(); + + SocketAddress firstAddress = new InetSocketAddress("127.0.0.1", firstServer.getLocalPort()); + SocketAddress secondAddress = new InetSocketAddress("127.0.0.1", secondServer.getLocalPort()); + + ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); + IoSessionInitiator initiator = createInitiator(new SocketAddress[]{firstAddress, secondAddress}, executor); + + try { + initiator.start(); + + assertTrue(firstConnection.await(5, TimeUnit.SECONDS)); + assertTrue(secondConnection.await(5, TimeUnit.SECONDS)); + } finally { + initiator.stop(); + executor.shutdownNow(); + } + } + } + + @Test + public void shouldResetSocketAddressAfterSuccessfulLogon() throws Exception { + try (ServerSocket firstServer = new ServerSocket(0); + ServerSocket secondServer = new ServerSocket(0)) { + + CountDownLatch firstConnection = new CountDownLatch(1); + CountDownLatch closeFirstConnection = new CountDownLatch(1); + CountDownLatch firstReconnect = new CountDownLatch(1); + CountDownLatch secondConnection = new CountDownLatch(1); + + Thread firstServerThread = new Thread(() -> { + try { + try (Socket socket = firstServer.accept()) { + firstConnection.countDown(); + closeFirstConnection.await(5, TimeUnit.SECONDS); + } + + try (Socket socket = firstServer.accept()) { + firstReconnect.countDown(); + } + } catch (Exception ignored) { + } + }); + + Thread secondServerThread = new Thread(() -> { + try (Socket socket = secondServer.accept()) { + secondConnection.countDown(); + } catch (Exception ignored) { + } + }); + + firstServerThread.start(); + secondServerThread.start(); + + Session session = mock(Session.class); + when(session.getLog()).thenReturn(mock(Log.class)); + when(session.isEnabled()).thenReturn(true); + when(session.isSessionTime()).thenReturn(true); + + ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); + + IoSessionInitiator initiator = createInitiator( + session, + new SocketAddress[]{ + new InetSocketAddress("127.0.0.1", firstServer.getLocalPort()), + new InetSocketAddress("127.0.0.1", secondServer.getLocalPort()) + }, + executor + ); + + try { + initiator.start(); + + assertTrue(firstConnection.await(5, TimeUnit.SECONDS)); + + ArgumentCaptor captor = ArgumentCaptor.forClass(SessionStateListener.class); + verify(session).addStateListener(captor.capture()); + captor.getValue().onLogon(session.getSessionID()); + closeFirstConnection.countDown(); + + assertTrue(firstReconnect.await(5, TimeUnit.SECONDS)); + assertEquals(1, secondConnection.getCount()); + } finally { + initiator.stop(); + executor.shutdownNow(); + } + } + } + private void assertProxyIsNotConfigured(SocketAddress socketAddress, String proxyType, int proxyPort) throws Exception { IoSessionInitiator initiator = createInitiator(socketAddress, proxyType, proxyPort); @@ -57,4 +180,70 @@ private IoSessionInitiator createInitiator(SocketAddress socketAddress, String p false, null, proxyType, "5", "127.0.0.1", proxyPort, null, null, null, null); } + + private IoSessionInitiator createInitiator( + SocketAddress[] socketAddresses, + ScheduledExecutorService executor) + throws Exception { + + Session session = mock(Session.class); + when(session.getLog()).thenReturn(mock(Log.class)); + when(session.isEnabled()).thenReturn(true); + when(session.isSessionTime()).thenReturn(true); + + return new IoSessionInitiator( + session, + socketAddresses, + null, + HostResolutionStrategy.WITHOUT_REVERSE_DNS, + 1, + new int[] { 1 }, + executor, + new SessionSettings(), + new NetworkingOptions(new Properties()), + mock(EventHandlingStrategy.class), + null, + false, + null, + null, + null, + null, + 0, + null, + null, + null, + null + ); + } + + private IoSessionInitiator createInitiator( + Session session, + SocketAddress[] socketAddresses, + ScheduledExecutorService executor) + throws Exception { + + return new IoSessionInitiator( + session, + socketAddresses, + null, + HostResolutionStrategy.WITHOUT_REVERSE_DNS, + 1, + new int[]{1}, + executor, + new SessionSettings(), + new NetworkingOptions(new Properties()), + mock(EventHandlingStrategy.class), + null, + false, + null, + null, + null, + null, + 0, + null, + null, + null, + null + ); + } }