From a9bf29cdb5519bce97a0a773bdd2c2af61f7f7ff Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 21 Oct 2026 14:53:55 +0000 Subject: [PATCH 1/5] Reuse initiator connectors after disconnects Co-authored-by: chrjohn <6644028+chrjohn@users.noreply.github.com> --- .../java/quickfix/mina/AbstractIoHandler.java | 14 +++++--------- .../java/quickfix/mina/IoSessionResponder.java | 1 - .../java/quickfix/mina/SessionConnector.java | 1 - .../mina/initiator/IoSessionInitiator.java | 18 ------------------ .../quickfix/mina/IoSessionResponderTest.java | 1 - 5 files changed, 5 insertions(+), 30 deletions(-) diff --git a/quickfixj-core/src/main/java/quickfix/mina/AbstractIoHandler.java b/quickfixj-core/src/main/java/quickfix/mina/AbstractIoHandler.java index ad4b3dae36..62859b4cc4 100644 --- a/quickfixj-core/src/main/java/quickfix/mina/AbstractIoHandler.java +++ b/quickfixj-core/src/main/java/quickfix/mina/AbstractIoHandler.java @@ -102,15 +102,11 @@ public void exceptionCaught(IoSession ioSession, Throwable cause) throws Excepti reason = cause.toString(); } if (disconnectNeeded) { - try { - if (quickFixSession != null) { - quickFixSession.disconnect(reason, true); - } else { - log.error(reason, cause); - ioSession.closeOnFlush(); - } - } finally { - ioSession.setAttribute(SessionConnector.QFJ_RESET_IO_CONNECTOR, Boolean.TRUE); + if (quickFixSession != null) { + quickFixSession.disconnect(reason, true); + } else { + log.error(reason, cause); + ioSession.closeOnFlush(); } } else { if (quickFixSession != null) { diff --git a/quickfixj-core/src/main/java/quickfix/mina/IoSessionResponder.java b/quickfixj-core/src/main/java/quickfix/mina/IoSessionResponder.java index 0a8a17fbc1..9aa173a79f 100644 --- a/quickfixj-core/src/main/java/quickfix/mina/IoSessionResponder.java +++ b/quickfixj-core/src/main/java/quickfix/mina/IoSessionResponder.java @@ -80,7 +80,6 @@ public void disconnect() { // close event from being processed by this thread (if // this thread is the MINA IO processor thread). ioSession.closeOnFlush(); - ioSession.setAttribute(SessionConnector.QFJ_RESET_IO_CONNECTOR, Boolean.TRUE); } @Override diff --git a/quickfixj-core/src/main/java/quickfix/mina/SessionConnector.java b/quickfixj-core/src/main/java/quickfix/mina/SessionConnector.java index 74cdbda99d..9f621b1961 100644 --- a/quickfixj-core/src/main/java/quickfix/mina/SessionConnector.java +++ b/quickfixj-core/src/main/java/quickfix/mina/SessionConnector.java @@ -63,7 +63,6 @@ public abstract class SessionConnector implements Connector { protected static final int DEFAULT_QUEUE_CAPACITY = 10000; public static final String SESSIONS_PROPERTY = "sessions"; public static final String QF_SESSION = "QF_SESSION"; - public static final String QFJ_RESET_IO_CONNECTOR = "QFJ_RESET_IO_CONNECTOR"; protected final Logger log = LoggerFactory.getLogger(getClass()); 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 84adf56e18..b11e4390d7 100644 --- a/quickfixj-core/src/main/java/quickfix/mina/initiator/IoSessionInitiator.java +++ b/quickfixj-core/src/main/java/quickfix/mina/initiator/IoSessionInitiator.java @@ -225,7 +225,6 @@ public String getSniHostName(SSLConfig sslConfig) { @Override public void run() { - resetIoConnector(); try { if (connectFuture == null) { if (shouldReconnect()) { @@ -352,23 +351,6 @@ public Session getFixSession() { return fixSession; } - private void resetIoConnector() { - if (ioSession != null && Boolean.TRUE.equals(ioSession.getAttribute(SessionConnector.QFJ_RESET_IO_CONNECTOR))) { - try { - setupIoConnector(); - if (connectFuture != null) { - connectFuture.cancel(); - } - connectFuture = null; - if (!ioSession.isClosing()) { - ioSession.closeNow(); - } - ioSession = null; - } catch (Throwable e) { - LogUtil.logThrowable(fixSession.getLog(), "Exception during resetIoConnector call", e); - } - } - } } synchronized void start() { diff --git a/quickfixj-core/src/test/java/quickfix/mina/IoSessionResponderTest.java b/quickfixj-core/src/test/java/quickfix/mina/IoSessionResponderTest.java index 4da872c1ae..e821c48228 100644 --- a/quickfixj-core/src/test/java/quickfix/mina/IoSessionResponderTest.java +++ b/quickfixj-core/src/test/java/quickfix/mina/IoSessionResponderTest.java @@ -135,7 +135,6 @@ public void testDisconnect() throws Exception { responder.disconnect(); verify(mockProtocolSession).closeOnFlush(); - verify(mockProtocolSession).setAttribute("QFJ_RESET_IO_CONNECTOR", Boolean.TRUE); verifyNoMoreInteractions(mockProtocolSession); } From 80da17f5a3bf8eecfbaf079f49a19b0d3250baa2 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 22 Oct 2026 05:03:32 +0000 Subject: [PATCH 2/5] Reset proxy connectors after disconnects Co-authored-by: chrjohn <6644028+chrjohn@users.noreply.github.com> --- .../java/quickfix/mina/AbstractIoHandler.java | 14 ++++++++----- .../quickfix/mina/IoSessionResponder.java | 1 + .../java/quickfix/mina/SessionConnector.java | 1 + .../mina/initiator/IoSessionInitiator.java | 20 +++++++++++++++++++ .../quickfix/mina/IoSessionResponderTest.java | 1 + 5 files changed, 32 insertions(+), 5 deletions(-) diff --git a/quickfixj-core/src/main/java/quickfix/mina/AbstractIoHandler.java b/quickfixj-core/src/main/java/quickfix/mina/AbstractIoHandler.java index 62859b4cc4..ad4b3dae36 100644 --- a/quickfixj-core/src/main/java/quickfix/mina/AbstractIoHandler.java +++ b/quickfixj-core/src/main/java/quickfix/mina/AbstractIoHandler.java @@ -102,11 +102,15 @@ public void exceptionCaught(IoSession ioSession, Throwable cause) throws Excepti reason = cause.toString(); } if (disconnectNeeded) { - if (quickFixSession != null) { - quickFixSession.disconnect(reason, true); - } else { - log.error(reason, cause); - ioSession.closeOnFlush(); + try { + if (quickFixSession != null) { + quickFixSession.disconnect(reason, true); + } else { + log.error(reason, cause); + ioSession.closeOnFlush(); + } + } finally { + ioSession.setAttribute(SessionConnector.QFJ_RESET_IO_CONNECTOR, Boolean.TRUE); } } else { if (quickFixSession != null) { diff --git a/quickfixj-core/src/main/java/quickfix/mina/IoSessionResponder.java b/quickfixj-core/src/main/java/quickfix/mina/IoSessionResponder.java index 9aa173a79f..0a8a17fbc1 100644 --- a/quickfixj-core/src/main/java/quickfix/mina/IoSessionResponder.java +++ b/quickfixj-core/src/main/java/quickfix/mina/IoSessionResponder.java @@ -80,6 +80,7 @@ public void disconnect() { // close event from being processed by this thread (if // this thread is the MINA IO processor thread). ioSession.closeOnFlush(); + ioSession.setAttribute(SessionConnector.QFJ_RESET_IO_CONNECTOR, Boolean.TRUE); } @Override diff --git a/quickfixj-core/src/main/java/quickfix/mina/SessionConnector.java b/quickfixj-core/src/main/java/quickfix/mina/SessionConnector.java index 9f621b1961..74cdbda99d 100644 --- a/quickfixj-core/src/main/java/quickfix/mina/SessionConnector.java +++ b/quickfixj-core/src/main/java/quickfix/mina/SessionConnector.java @@ -63,6 +63,7 @@ public abstract class SessionConnector implements Connector { protected static final int DEFAULT_QUEUE_CAPACITY = 10000; public static final String SESSIONS_PROPERTY = "sessions"; public static final String QF_SESSION = "QF_SESSION"; + public static final String QFJ_RESET_IO_CONNECTOR = "QFJ_RESET_IO_CONNECTOR"; protected final Logger log = LoggerFactory.getLogger(getClass()); 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 b11e4390d7..ea4ac56b4d 100644 --- a/quickfixj-core/src/main/java/quickfix/mina/initiator/IoSessionInitiator.java +++ b/quickfixj-core/src/main/java/quickfix/mina/initiator/IoSessionInitiator.java @@ -225,6 +225,7 @@ public String getSniHostName(SSLConfig sslConfig) { @Override public void run() { + resetProxyConnector(); try { if (connectFuture == null) { if (shouldReconnect()) { @@ -351,6 +352,25 @@ public Session getFixSession() { return fixSession; } + private void resetProxyConnector() { + if (ioConnector instanceof ProxyConnector && ioSession != null + && Boolean.TRUE.equals(ioSession.getAttribute(SessionConnector.QFJ_RESET_IO_CONNECTOR))) { + try { + setupIoConnector(); + if (connectFuture != null) { + connectFuture.cancel(); + } + connectFuture = null; + if (!ioSession.isClosing()) { + ioSession.closeNow(); + } + ioSession = null; + } catch (Throwable e) { + LogUtil.logThrowable(fixSession.getLog(), "Exception during resetProxyConnector call", e); + } + } + } + } synchronized void start() { diff --git a/quickfixj-core/src/test/java/quickfix/mina/IoSessionResponderTest.java b/quickfixj-core/src/test/java/quickfix/mina/IoSessionResponderTest.java index e821c48228..4da872c1ae 100644 --- a/quickfixj-core/src/test/java/quickfix/mina/IoSessionResponderTest.java +++ b/quickfixj-core/src/test/java/quickfix/mina/IoSessionResponderTest.java @@ -135,6 +135,7 @@ public void testDisconnect() throws Exception { responder.disconnect(); verify(mockProtocolSession).closeOnFlush(); + verify(mockProtocolSession).setAttribute("QFJ_RESET_IO_CONNECTOR", Boolean.TRUE); verifyNoMoreInteractions(mockProtocolSession); } From aba8443d907e1cf4b2769e102d81c0e1b489de30 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 22 Oct 2026 08:27:59 +0000 Subject: [PATCH 3/5] Add reconnect connector reuse regression test Co-authored-by: chrjohn <6644028+chrjohn@users.noreply.github.com> --- .../java/quickfix/SocketInitiatorTest.java | 114 ++++++++++++++++++ 1 file changed, 114 insertions(+) diff --git a/quickfixj-core/src/test/java/quickfix/SocketInitiatorTest.java b/quickfixj-core/src/test/java/quickfix/SocketInitiatorTest.java index 89a0de8637..7bc6816b1c 100644 --- a/quickfixj-core/src/test/java/quickfix/SocketInitiatorTest.java +++ b/quickfixj-core/src/test/java/quickfix/SocketInitiatorTest.java @@ -47,7 +47,9 @@ import java.net.Socket; import java.net.SocketAddress; import java.util.HashMap; +import java.util.Set; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; @@ -134,6 +136,36 @@ public void testLogonAfterServerDisconnect() throws Exception { assertTrue("Acceptor sessionCreated not called", serverThread.wasSessionCreatedCalled()); } + @Test + public void testReconnectAfterServerDisconnectReusesIoConnector() throws Exception { + final int reconnects = 3; + DisconnectingServer server = new DisconnectingServer(reconnects); + server.start(); + + SessionID clientSessionID = new SessionID(FixVersions.BEGINSTRING_FIX42, "TW", "ISLD"); + SessionSettings settings = getClientSessionSettings(clientSessionID, server.getPort()); + settings.setString(clientSessionID, "SocketConnectProtocol", ProtocolFactory.getTypeString(ProtocolFactory.SOCKET)); + settings.setString(clientSessionID, "SocketConnectHost", "127.0.0.1"); + settings.setString(clientSessionID, "ReconnectInterval", "1"); + Initiator initiator = new SocketInitiator(new ApplicationAdapter(), new MemoryStoreFactory(), settings, + new DefaultMessageFactory()); + ConnectorThreadMonitor connectorThreadMonitor = new ConnectorThreadMonitor(); + connectorThreadMonitor.start(); + + try { + initiator.start(); + assertTrue("Initiator did not reconnect after server disconnects", + server.awaitConnections(15, TimeUnit.SECONDS)); + } finally { + initiator.stop(); + connectorThreadMonitor.stop(); + server.stop(); + } + + assertEquals("Direct initiator reconnects should reuse the same IoConnector", 1, + connectorThreadMonitor.getConnectorThreadNames().size()); + } + @Test public void testInitiatorStop() throws Exception { int freePort = AvailablePortFinder.getNextAvailable(); @@ -654,6 +686,88 @@ public String toString() { } } + private static class ConnectorThreadMonitor { + private static final String CONNECTOR_THREAD_PREFIX = "NioSocketConnector-"; + + private final Set connectorThreadNames = ConcurrentHashMap.newKeySet(); + private final AtomicBoolean running = new AtomicBoolean(); + private Thread thread; + + public void start() { + running.set(true); + thread = new Thread(() -> { + while (running.get()) { + for (Thread activeThread : Thread.getAllStackTraces().keySet()) { + String threadName = activeThread.getName(); + if (threadName.startsWith(CONNECTOR_THREAD_PREFIX)) { + connectorThreadNames.add(threadName); + } + } + try { + Thread.sleep(10); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + } + }); + thread.start(); + } + + public void stop() throws InterruptedException { + running.set(false); + thread.interrupt(); + thread.join(); + } + + public Set getConnectorThreadNames() { + return connectorThreadNames; + } + } + + private static class DisconnectingServer { + private final ServerSocket serverSocket; + private final CountDownLatch connections; + private final AtomicBoolean running = new AtomicBoolean(); + private Thread thread; + + DisconnectingServer(int expectedConnections) throws IOException { + serverSocket = new ServerSocket(0); + connections = new CountDownLatch(expectedConnections); + } + + public void start() { + running.set(true); + thread = new Thread(() -> { + while (running.get()) { + try (Socket socket = serverSocket.accept()) { + socket.getInputStream().read(); + connections.countDown(); + } catch (IOException e) { + if (running.get()) { + throw new RuntimeException(e); + } + } + } + }); + thread.start(); + } + + public int getPort() { + return serverSocket.getLocalPort(); + } + + public boolean awaitConnections(long timeout, TimeUnit unit) throws InterruptedException { + return connections.await(timeout, unit); + } + + public void stop() throws IOException, InterruptedException { + running.set(false); + serverSocket.close(); + thread.join(); + } + } + private class ServerThread extends Thread { private final ATServer server; private final WriteCounter writeCounter = new WriteCounter("acceptor"); From 2bf8959e2619c26821899cc227297535d276e5a2 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 22 Oct 2026 10:32:37 +0000 Subject: [PATCH 4/5] Count connector thread delta in reconnect test Co-authored-by: chrjohn <6644028+chrjohn@users.noreply.github.com> --- .../java/quickfix/SocketInitiatorTest.java | 55 ++++--------------- 1 file changed, 11 insertions(+), 44 deletions(-) diff --git a/quickfixj-core/src/test/java/quickfix/SocketInitiatorTest.java b/quickfixj-core/src/test/java/quickfix/SocketInitiatorTest.java index 7bc6816b1c..d64d2b596f 100644 --- a/quickfixj-core/src/test/java/quickfix/SocketInitiatorTest.java +++ b/quickfixj-core/src/test/java/quickfix/SocketInitiatorTest.java @@ -47,9 +47,7 @@ import java.net.Socket; import java.net.SocketAddress; import java.util.HashMap; -import java.util.Set; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; @@ -139,6 +137,8 @@ public void testLogonAfterServerDisconnect() throws Exception { @Test public void testReconnectAfterServerDisconnectReusesIoConnector() throws Exception { final int reconnects = 3; + int connectorThreadsBefore = getNioSocketConnectorThreadCount(); + log.info("NioSocketConnector threads before test: {}", connectorThreadsBefore); DisconnectingServer server = new DisconnectingServer(reconnects); server.start(); @@ -149,21 +149,17 @@ public void testReconnectAfterServerDisconnectReusesIoConnector() throws Excepti settings.setString(clientSessionID, "ReconnectInterval", "1"); Initiator initiator = new SocketInitiator(new ApplicationAdapter(), new MemoryStoreFactory(), settings, new DefaultMessageFactory()); - ConnectorThreadMonitor connectorThreadMonitor = new ConnectorThreadMonitor(); - connectorThreadMonitor.start(); try { initiator.start(); assertTrue("Initiator did not reconnect after server disconnects", server.awaitConnections(15, TimeUnit.SECONDS)); + assertEquals("Direct initiator reconnects should reuse the same IoConnector", + connectorThreadsBefore + 1, getNioSocketConnectorThreadCount()); } finally { initiator.stop(); - connectorThreadMonitor.stop(); server.stop(); } - - assertEquals("Direct initiator reconnects should reuse the same IoConnector", 1, - connectorThreadMonitor.getConnectorThreadNames().size()); } @Test @@ -686,43 +682,14 @@ public String toString() { } } - private static class ConnectorThreadMonitor { - private static final String CONNECTOR_THREAD_PREFIX = "NioSocketConnector-"; - - private final Set connectorThreadNames = ConcurrentHashMap.newKeySet(); - private final AtomicBoolean running = new AtomicBoolean(); - private Thread thread; - - public void start() { - running.set(true); - thread = new Thread(() -> { - while (running.get()) { - for (Thread activeThread : Thread.getAllStackTraces().keySet()) { - String threadName = activeThread.getName(); - if (threadName.startsWith(CONNECTOR_THREAD_PREFIX)) { - connectorThreadNames.add(threadName); - } - } - try { - Thread.sleep(10); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return; - } - } - }); - thread.start(); - } - - public void stop() throws InterruptedException { - running.set(false); - thread.interrupt(); - thread.join(); - } - - public Set getConnectorThreadNames() { - return connectorThreadNames; + private int getNioSocketConnectorThreadCount() { + int count = 0; + for (Thread thread : Thread.getAllStackTraces().keySet()) { + if (thread.getName().startsWith("NioSocketConnector-")) { + count++; + } } + return count; } private static class DisconnectingServer { From df5721805b7764a186a3d6b7dd9fce3d64c2c2bb Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 23 Oct 2026 18:57:08 +0000 Subject: [PATCH 5/5] Assert connector identity across reconnects Co-authored-by: chrjohn <6644028+chrjohn@users.noreply.github.com> --- .../java/quickfix/SocketInitiatorTest.java | 34 ++++++++++++------- 1 file changed, 22 insertions(+), 12 deletions(-) diff --git a/quickfixj-core/src/test/java/quickfix/SocketInitiatorTest.java b/quickfixj-core/src/test/java/quickfix/SocketInitiatorTest.java index d64d2b596f..886098eec4 100644 --- a/quickfixj-core/src/test/java/quickfix/SocketInitiatorTest.java +++ b/quickfixj-core/src/test/java/quickfix/SocketInitiatorTest.java @@ -31,7 +31,9 @@ import quickfix.field.MsgType; import quickfix.mina.ProtocolFactory; import quickfix.mina.SingleThreadedEventHandlingStrategy; +import quickfix.mina.initiator.AbstractSocketInitiator; import quickfix.mina.initiator.ConnectException; +import quickfix.mina.initiator.IoSessionInitiator; import quickfix.mina.ssl.SSLSupport; import quickfix.test.acceptance.ATServer; import quickfix.test.util.StackTraceUtil; @@ -39,6 +41,7 @@ import java.io.File; import java.io.IOException; import java.io.InputStream; +import java.lang.reflect.Field; import java.lang.management.ManagementFactory; import java.lang.management.ThreadInfo; import java.lang.management.ThreadMXBean; @@ -47,6 +50,7 @@ import java.net.Socket; import java.net.SocketAddress; import java.util.HashMap; +import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -59,6 +63,7 @@ import static junit.framework.TestCase.assertNotNull; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; @@ -137,8 +142,6 @@ public void testLogonAfterServerDisconnect() throws Exception { @Test public void testReconnectAfterServerDisconnectReusesIoConnector() throws Exception { final int reconnects = 3; - int connectorThreadsBefore = getNioSocketConnectorThreadCount(); - log.info("NioSocketConnector threads before test: {}", connectorThreadsBefore); DisconnectingServer server = new DisconnectingServer(reconnects); server.start(); @@ -152,10 +155,11 @@ public void testReconnectAfterServerDisconnectReusesIoConnector() throws Excepti try { initiator.start(); + Object connectorBeforeReconnects = getIoConnector((AbstractSocketInitiator) initiator); assertTrue("Initiator did not reconnect after server disconnects", server.awaitConnections(15, TimeUnit.SECONDS)); - assertEquals("Direct initiator reconnects should reuse the same IoConnector", - connectorThreadsBefore + 1, getNioSocketConnectorThreadCount()); + assertSame("Direct initiator reconnects should reuse the same IoConnector", + connectorBeforeReconnects, getIoConnector((AbstractSocketInitiator) initiator)); } finally { initiator.stop(); server.stop(); @@ -682,14 +686,20 @@ public String toString() { } } - private int getNioSocketConnectorThreadCount() { - int count = 0; - for (Thread thread : Thread.getAllStackTraces().keySet()) { - if (thread.getName().startsWith("NioSocketConnector-")) { - count++; - } - } - return count; + @SuppressWarnings("unchecked") + private Object getIoConnector(AbstractSocketInitiator initiator) throws ReflectiveOperationException { + Field initiatorsField = AbstractSocketInitiator.class.getDeclaredField("initiators"); + initiatorsField.setAccessible(true); + Set initiators = (Set) initiatorsField.get(initiator); + assertEquals(1, initiators.size()); + + Field reconnectTaskField = IoSessionInitiator.class.getDeclaredField("reconnectTask"); + reconnectTaskField.setAccessible(true); + Object reconnectTask = reconnectTaskField.get(initiators.iterator().next()); + + Field ioConnectorField = reconnectTask.getClass().getDeclaredField("ioConnector"); + ioConnectorField.setAccessible(true); + return ioConnectorField.get(reconnectTask); } private static class DisconnectingServer {