Skip to content

Fix socket address failover before successful logon #1324

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
elliotoakley wants to merge 1 commit into
base: master
Choose a base branch
from
+210 −3
Open

Fix socket address failover before successful logon #1324

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 21 additions & 3 deletions quickfixj-core/src/main/java/quickfix/mina/initiator/IoSessionInitiator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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));
}

Expand All @@ -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;

Expand Down Expand Up @@ -225,6 +235,10 @@ public String getSniHostName(SSLConfig sslConfig) {

@Override
public void run() {
if (socketAddressIndexResetRequested) {
nextSocketAddressIndex = 0;
socketAddressIndexResetRequested = false;
}
resetIoConnector();
try {
if (connectFuture == null) {
Expand Down Expand Up @@ -261,7 +275,6 @@ private void pollConnectFuture() {
if (connectFuture.getSession() != null) {
ioSession = connectFuture.getSession();
connectionFailureCount = 0;
nextSocketAddressIndex = 0;
lastConnectTime = System.currentTimeMillis();
connectFuture = null;
} else {
Expand Down Expand Up @@ -369,22 +382,27 @@ private void resetIoConnector() {
}
}
}

private void resetSocketAddressIndex() {
socketAddressIndexResetRequested = true;
}
}

synchronized void start() {
if (reconnectFuture == null) {
// 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);
}
}

synchronized void stop() {
if (reconnectFuture != null) {
reconnectFuture.cancel(true);
reconnectFuture = null;
reconnectTask.getFixSession().removeStateListener(stateListener);
}
SessionConnector.closeManagedSessionsAndDispose(reconnectTask.ioConnector, true, log);
}
Expand Down
189 changes: 189 additions & 0 deletions quickfixj-core/src/test/java/quickfix/mina/initiator/IoSessionInitiatorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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 {

Expand All @@ -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<SessionStateListener> 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);
Expand All @@ -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
);
}
}
Loading