QFJ-942: Observing hangs in dispose() on stop()

quickfixj-core/src/main/java/quickfix/mina/QueueTracker.java

@@ -1,7 +1,6 @@
 package quickfix.mina;
 
 import java.util.Collection;
-import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.TimeUnit;
 
 interface QueueTracker<E> {

quickfixj-core/src/main/java/quickfix/mina/SessionConnector.java

@@ -50,6 +50,8 @@
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
+import org.apache.mina.core.future.CloseFuture;
+import org.apache.mina.core.service.IoService;
 
 /**
  * An abstract base class for acceptors and initiators. Provides support for common functionality and also serves as an
@@ -423,4 +425,33 @@ public void setIoFilterChainBuilder(IoFilterChainBuilder ioFilterChainBuilder) {
     protected IoFilterChainBuilder getIoFilterChainBuilder() {
         return ioFilterChainBuilder;
     }
+
+    /**
+     * Closes all managed sessions of an Initiator/Acceptor.
+     *
+     * @param ioService Acceptor or Initiator implementation
+     * @param awaitTermination whether to wait for underlying ExecutorService to terminate
+     * @param logger used for logging WARNING when IoSession could not be closed
+     */
+    public static void closeManagedSessionsAndDispose(IoService ioService, boolean awaitTermination, Logger logger) {
+        Map<Long, IoSession> managedSessions = ioService.getManagedSessions();
+        for (IoSession ioSession : managedSessions.values()) {
+            if (!ioSession.isClosing()) {
+                CloseFuture closeFuture = ioSession.closeNow();
+                boolean completed = false;
+                try {
+                    completed = closeFuture.await(1000, TimeUnit.MILLISECONDS);
+                } catch (InterruptedException ex) {
+                    Thread.currentThread().interrupt();
+                }
+                if (!completed) {
+                    logger.warn("Could not close IoSession {}", ioSession);
+                }
+            }
+        }
+        if (!ioService.isDisposing()) {
+            ioService.dispose(awaitTermination);
+        }
+    }
+
 }

quickfixj-core/src/main/java/quickfix/mina/acceptor/AbstractSocketAcceptor.java

@@ -248,7 +248,7 @@ protected void stopAcceptingConnections() throws ConfigError {
             IoAcceptor ioAcceptor = ioIt.next();
             SocketAddress localAddress = ioAcceptor.getLocalAddress();
             ioAcceptor.unbind();
-            ioAcceptor.dispose(true);
+            closeManagedSessionsAndDispose(ioAcceptor, true, log);
             log.info("No longer accepting connections on {}", localAddress);
             ioIt.remove();
         }

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

@@ -50,11 +50,14 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class IoSessionInitiator {
     private final static long CONNECT_POLL_TIMEOUT = 2000L;
     private final ScheduledExecutorService executor;
     private final ConnectTask reconnectTask;
+    private final Logger log = LoggerFactory.getLogger(getClass());
 
     private Future<?> reconnectFuture;
 
@@ -74,7 +77,7 @@ public IoSessionInitiator(Session fixSession, SocketAddress[] socketAddresses,
             reconnectTask = new ConnectTask(sslEnabled, socketAddresses, localAddress,
                     userIoFilterChainBuilder, fixSession, reconnectIntervalInMillis,
                     networkingOptions, eventHandlingStrategy, sslConfig,
-                    proxyType, proxyVersion, proxyHost, proxyPort, proxyUser, proxyPassword, proxyDomain, proxyWorkstation);
+                    proxyType, proxyVersion, proxyHost, proxyPort, proxyUser, proxyPassword, proxyDomain, proxyWorkstation, log);
         } catch (GeneralSecurityException e) {
             throw new ConfigError(e);
         }
@@ -93,6 +96,7 @@ private static class ConnectTask implements Runnable {
         private final NetworkingOptions networkingOptions;
         private final EventHandlingStrategy eventHandlingStrategy;
         private final SSLConfig sslConfig;
+        private final Logger log;
 
         private IoSession ioSession;
         private long lastReconnectAttemptTime;
@@ -116,7 +120,7 @@ public ConnectTask(boolean sslEnabled, SocketAddress[] socketAddresses,
                 NetworkingOptions networkingOptions, EventHandlingStrategy eventHandlingStrategy, SSLConfig sslConfig,
                 String proxyType, String proxyVersion, String proxyHost,
                 int proxyPort, String proxyUser, String proxyPassword, String proxyDomain,
-                String proxyWorkstation) throws ConfigError, GeneralSecurityException {
+                String proxyWorkstation, Logger log) throws ConfigError, GeneralSecurityException {
             this.sslEnabled = sslEnabled;
             this.socketAddresses = socketAddresses;
             this.localAddress = localAddress;
@@ -126,6 +130,7 @@ public ConnectTask(boolean sslEnabled, SocketAddress[] socketAddresses,
             this.networkingOptions = networkingOptions;
             this.eventHandlingStrategy = eventHandlingStrategy;
             this.sslConfig = sslConfig;
+            this.log = log;
 
             this.proxyType = proxyType;
             this.proxyVersion = proxyVersion;
@@ -173,7 +178,7 @@ private void setupIoConnector() throws ConfigError, GeneralSecurityException {
             }
 
             if (ioConnector != null) {
-                ioConnector.dispose();
+                SessionConnector.closeManagedSessionsAndDispose(ioConnector, true, log);
             }
             ioConnector = newConnector;
         }
@@ -338,6 +343,9 @@ private void resetIoConnector() {
                         connectFuture.cancel();
                     }
                     connectFuture = null;
+                    if (!ioSession.isClosing()) {
+                        ioSession.closeNow();
+                    }
                     ioSession = null;
                 } catch (Throwable e) {
                     LogUtil.logThrowable(fixSession.getLog(), "Exception during resetIoConnector call", e);
@@ -361,7 +369,6 @@ synchronized void stop() {
             reconnectFuture.cancel(true);
             reconnectFuture = null;
         }
-        // QFJ-849: clean up resources of MINA connector
-        reconnectTask.ioConnector.dispose();
+        SessionConnector.closeManagedSessionsAndDispose(reconnectTask.ioConnector, true, log);
     }
 }

quickfixj-core/src/test/java/quickfix/SocketInitiatorTest.java

@@ -376,10 +376,11 @@ private void doTestOfRestart(SessionID clientSessionID, ClientApplication client
                 Session clientSession = Session.lookupSession(clientSessionID);
                 assertLoggedOn(clientApplication, clientSession);
 
+                clientApplication.setUpLogoutExpectation();
                 initiator.stop();
-                assertFalse(clientSession.isLoggedOn());
+                assertLoggedOut(clientApplication, clientSession);
                 assertFalse(initiator.getSessions().contains(clientSessionID));
-                assertTrue(initiator.getSessions().size() == 0);
+                assertTrue(initiator.getSessions().isEmpty());
                 if (messageLog != null) {
                     messageLogLength = messageLog.length();
                     assertTrue(messageLog.length() > 0);
@@ -482,9 +483,6 @@ private void assertLoggedOn(ClientApplication clientApplication, Session clientS
         }
 
         final boolean await = clientApplication.logonLatch.await(20, TimeUnit.SECONDS); 
-        if (!await) {
-            ReflectionUtil.dumpStackTraces();
-        }
         assertTrue("Expected logon did not occur", await); 
         assertTrue("client session not logged in", clientSession.isLoggedOn());
     }

quickfixj-core/src/test/java/quickfix/test/acceptance/TestConnection.java

@@ -48,6 +48,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
+import quickfix.mina.SessionConnector;
 
 public class TestConnection {
     private static final HashMap<String, IoConnector> connectors = new HashMap<>();
@@ -85,27 +86,20 @@ public void connect(int clientId, int transportType, int port)
             throws IOException {
         IoConnector connector = connectors.get(Integer.toString(clientId));
         if (connector != null) {
-            log.info("Disposing connector for clientId " + clientId);
-            connector.dispose();
+            SessionConnector.closeManagedSessionsAndDispose(connector, true, log);
         }
 
-        if (transportType == ProtocolFactory.SOCKET) {
-            connector = new NioSocketConnector();
-        } else if (transportType == ProtocolFactory.VM_PIPE) {
-            connector = new VmPipeConnector();
-        } else {
-            throw new RuntimeException("Unsupported transport type: " + transportType);
-        }
-        connectors.put(Integer.toString(clientId), connector);
-
         SocketAddress address;
         if (transportType == ProtocolFactory.SOCKET) {
+            connector = new NioSocketConnector();
             address = new InetSocketAddress("localhost", port);
         } else if (transportType == ProtocolFactory.VM_PIPE) {
+            connector = new VmPipeConnector();
             address = new VmPipeAddress(port);
         } else {
             throw new RuntimeException("Unsupported transport type: " + transportType);
         }
+        connectors.put(Integer.toString(clientId), connector);
 
         TestIoHandler testIoHandler = new TestIoHandler();
         synchronized (ioHandlers) {