Merge pull request #638 from AndreyNudko/socket-connect-timeout

Introduce `SocketConnectTimeout` configuration option

Author: chrjohn

quickfixj-core/src/main/doc/usermanual/usage/configuration.html

@@ -509,6 +509,13 @@ <H3>QuickFIX Settings</H3>
     <TD> valid IP address in the format of x.x.x.x or a domain name </TD>
     <TD>&nbsp; </TD>
   </TR>
+  <TR ALIGN="left" VALIGN="middle">
+    <TD> <I>SocketConnectTimeout</I> </TD>
+
+    <TD> Connection timeout in seconds. Only used with a SocketInitiator </TD>
+    <TD> positive integer </TD>
+    <TD> 60 </TD>
+  </TR>
   <TR ALIGN="left" VALIGN="middle">
 
     <TD> <I>SocketConnectProtocol</I> </TD>

quickfixj-core/src/main/java/quickfix/Initiator.java

@@ -53,6 +53,14 @@ public interface Initiator extends Connector {
      */
     String SETTING_SOCKET_CONNECT_PORT = "SocketConnectPort";
 
+    /**
+     * Initiator setting for connection timeout. Only valid when session connection
+     * type is "initiator".
+     *
+     * @see quickfix.SessionFactory#SETTING_CONNECTION_TYPE
+     */
+    String SETTING_SOCKET_CONNECT_TIMEOUT = "SocketConnectTimeout";
+
     /**
      * Initiator setting for local/bind host. Only valid when session connection
      * type is "initiator".

quickfixj-core/src/main/java/quickfix/mina/initiator/AbstractSocketInitiator.java

@@ -124,8 +124,10 @@ private void createInitiator(final Session session, final boolean continueInitOn
             throw new ConfigError("Must specify at least one socket address");
         }
 
-        SocketAddress localAddress = getLocalAddress(settings, session);
+        // 1 minute by default, matches MINA
+        int connectTimeout = getSettings().getIntOrDefault(sessionID, Initiator.SETTING_SOCKET_CONNECT_TIMEOUT, 60);
 
+        SocketAddress localAddress = getLocalAddress(settings, session);
         final NetworkingOptions networkingOptions = new NetworkingOptions(getSettings()
                 .getSessionProperties(sessionID, true));
 
@@ -175,7 +177,7 @@ && getSettings().isSetting(sessionID, Initiator.SETTING_PROXY_DOMAIN)) {
         ScheduledExecutorService scheduledExecutorService = (scheduledReconnectExecutor != null ? scheduledReconnectExecutor : getScheduledExecutorService());
         try {
             final IoSessionInitiator ioSessionInitiator = new IoSessionInitiator(session,
-                    socketAddresses, localAddress, reconnectingIntervals,
+                    socketAddresses, localAddress, connectTimeout, reconnectingIntervals,
                     scheduledExecutorService, settings, networkingOptions,
                     getEventHandlingStrategy(), getIoFilterChainBuilder(), sslEnabled, sslConfig,
                     proxyType, proxyVersion, proxyHost, proxyPort, proxyUser, proxyPassword, proxyDomain, proxyWorkstation);

quickfixj-core/src/main/java/quickfix/mina/initiator/IoSessionInitiator.java

@@ -65,20 +65,23 @@ public class IoSessionInitiator {
     private Future<?> reconnectFuture;
 
     public IoSessionInitiator(Session fixSession, SocketAddress[] socketAddresses,
-            SocketAddress localAddress, int[] reconnectIntervalInSeconds,
+            SocketAddress localAddress, int connectTimeout, int[] reconnectIntervalInSeconds,
             ScheduledExecutorService executor, SessionSettings sessionSettings, NetworkingOptions networkingOptions,
             EventHandlingStrategy eventHandlingStrategy,
             IoFilterChainBuilder userIoFilterChainBuilder, boolean sslEnabled, SSLConfig sslConfig,
             String proxyType, String proxyVersion, String proxyHost, int proxyPort,
             String proxyUser, String proxyPassword, String proxyDomain, String proxyWorkstation) throws ConfigError {
         this.executor = executor;
+
+        final long connectTimeoutMillis = connectTimeout * 1000L;
         final long[] reconnectIntervalInMillis = new long[reconnectIntervalInSeconds.length];
         for (int ii = 0; ii != reconnectIntervalInSeconds.length; ++ii) {
             reconnectIntervalInMillis[ii] = reconnectIntervalInSeconds[ii] * 1000L;
         }
+
         try {
             reconnectTask = new ConnectTask(sslEnabled, socketAddresses, localAddress,
-                    userIoFilterChainBuilder, fixSession, reconnectIntervalInMillis,
+                    userIoFilterChainBuilder, fixSession, connectTimeoutMillis, reconnectIntervalInMillis,
                     sessionSettings, networkingOptions, eventHandlingStrategy, sslConfig,
                     proxyType, proxyVersion, proxyHost, proxyPort, proxyUser, proxyPassword, proxyDomain, proxyWorkstation, log);
         } catch (GeneralSecurityException e) {
@@ -95,6 +98,7 @@ private static class ConnectTask implements Runnable {
         private final IoFilterChainBuilder userIoFilterChainBuilder;
         private IoConnector ioConnector;
         private final Session fixSession;
+        private final long connectTimeoutMillis;
         private final long[] reconnectIntervalInMillis;
         private final SessionSettings sessionSettings;
         private final NetworkingOptions networkingOptions;
@@ -120,7 +124,7 @@ private static class ConnectTask implements Runnable {
 
         public ConnectTask(boolean sslEnabled, SocketAddress[] socketAddresses,
                 SocketAddress localAddress, IoFilterChainBuilder userIoFilterChainBuilder,
-                Session fixSession, long[] reconnectIntervalInMillis,
+                Session fixSession, long connectTimeoutMillis, long[] reconnectIntervalInMillis,
                 SessionSettings sessionSettings, NetworkingOptions networkingOptions, EventHandlingStrategy eventHandlingStrategy, SSLConfig sslConfig,
                 String proxyType, String proxyVersion, String proxyHost,
                 int proxyPort, String proxyUser, String proxyPassword, String proxyDomain,
@@ -130,6 +134,7 @@ public ConnectTask(boolean sslEnabled, SocketAddress[] socketAddresses,
             this.localAddress = localAddress;
             this.userIoFilterChainBuilder = userIoFilterChainBuilder;
             this.fixSession = fixSession;
+            this.connectTimeoutMillis = connectTimeoutMillis;
             this.reconnectIntervalInMillis = reconnectIntervalInMillis;
             this.sessionSettings = sessionSettings;
             this.networkingOptions = networkingOptions;
@@ -161,9 +166,9 @@ private void setupIoConnector() throws ConfigError, GeneralSecurityException {
 
             ioFilterChainBuilder.addLast(FIXProtocolCodecFactory.FILTER_NAME, new ProtocolCodecFilter(new FIXProtocolCodecFactory()));
 
-            IoConnector newConnector;
-            newConnector = ProtocolFactory.createIoConnector(socketAddresses[nextSocketAddressIndex]);
+            IoConnector newConnector = ProtocolFactory.createIoConnector(socketAddresses[nextSocketAddressIndex]);
             networkingOptions.apply(newConnector);
+            newConnector.setConnectTimeoutMillis(connectTimeoutMillis);
             newConnector.setHandler(new InitiatorIoHandler(fixSession, sessionSettings, networkingOptions, eventHandlingStrategy));
             newConnector.setFilterChainBuilder(ioFilterChainBuilder);