Interrupt current thread where applicable. (#421)

chrjohn · web-flow · commit 6d2edfe9edab · 2021-10-12T14:26:30.000+02:00
* Added interruption of current thread on InterruptedException.
diff --git a/quickfixj-core/src/main/java/quickfix/mina/SessionConnector.java b/quickfixj-core/src/main/java/quickfix/mina/SessionConnector.java
@@ -272,6 +272,7 @@ protected void waitForLogout() {
                 Thread.sleep(100L);
             } catch (InterruptedException e) {
                 log.error(e.getMessage(), e);
+                Thread.currentThread().interrupt();
             }
             final long elapsed = System.currentTimeMillis() - start;
             Iterator<Session> sessionItr = loggedOnSessions.iterator();
@@ -375,6 +376,7 @@ public void run() {
             try {
                 delegate.await();
             } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
             }
         }
 
@@ -449,9 +451,10 @@ public static void closeManagedSessionsAndDispose(IoService ioService, boolean a
                     completed = closeFuture.await(1000, TimeUnit.MILLISECONDS);
                 } catch (InterruptedException ex) {
                     Thread.currentThread().interrupt();
-                }
-                if (!completed) {
-                    logger.warn("Could not close IoSession {}", ioSession);
+                } finally {
+                    if (!completed) {
+                        logger.warn("Could not close IoSession {}", ioSession);
+                    }
                 }
             }
         }
diff --git a/quickfixj-core/src/main/java/quickfix/mina/SingleThreadedEventHandlingStrategy.java b/quickfixj-core/src/main/java/quickfix/mina/SingleThreadedEventHandlingStrategy.java
@@ -76,7 +76,7 @@ public void onMessage(Session quickfixSession, Message message) {
             queueTracker.put(new SessionMessageEvent(quickfixSession, message));
         } catch (InterruptedException e) {
             isStopped = true;
-            throw new RuntimeException(e);
+            Thread.currentThread().interrupt();
         }
     }
 
@@ -196,7 +196,8 @@ public void stopHandlingMessages(boolean join) {
             try {
                 messageProcessingThread.join();
             } catch (InterruptedException e) {
-                sessionConnector.log.error("{} interrupted.", MESSAGE_PROCESSOR_THREAD_NAME);
+                sessionConnector.log.warn("{} interrupted.", MESSAGE_PROCESSOR_THREAD_NAME);
+                Thread.currentThread().interrupt();
             }
         }
     }
diff --git a/quickfixj-core/src/main/java/quickfix/mina/ThreadPerSessionEventHandlingStrategy.java b/quickfixj-core/src/main/java/quickfix/mina/ThreadPerSessionEventHandlingStrategy.java
@@ -203,6 +203,7 @@ public void enqueue(Message message) {
                 queueTracker.put(message);
             } catch (final InterruptedException e) {
                 quickfixSession.getLog().onErrorEvent(e.toString());
+                Thread.currentThread().interrupt();
             }
         }
 
@@ -227,6 +228,7 @@ void doRun() {
                     LogUtil.logThrowable(quickfixSession.getSessionID(),
                             "Message dispatcher interrupted", e);
                     stopping = true;
+                    Thread.currentThread().interrupt();
                 } catch (final Throwable e) {
                     LogUtil.logThrowable(quickfixSession.getSessionID(),
                             "Error during message processing", e);
diff --git a/quickfixj-core/src/test/java/quickfix/MultiAcceptorTest.java b/quickfixj-core/src/test/java/quickfix/MultiAcceptorTest.java
@@ -206,6 +206,7 @@ public void waitForLogon() {
             try {
                 logonLatch.await(20, TimeUnit.SECONDS);
             } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
                 fail(e.getMessage());
             }
         }
@@ -220,6 +221,7 @@ public synchronized void waitForMessages() {
                     fail("Timed out waiting for message");
                 }
             } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
                 fail(e.getMessage());
             }
         }
diff --git a/quickfixj-core/src/test/java/quickfix/SessionDisconnectConcurrentlyTest.java b/quickfixj-core/src/test/java/quickfix/SessionDisconnectConcurrentlyTest.java
@@ -134,6 +134,7 @@ public void waitForLogon() {
             try {
                 logonLatch.await(10, TimeUnit.SECONDS);
             } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
                 fail(e.getMessage());
             }
         }
@@ -148,6 +149,7 @@ public synchronized void waitForMessages() {
                     fail("Timed out waiting for message");
                 }
             } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
                 fail(e.getMessage());
             }
         }
@@ -225,7 +227,7 @@ public void run() {
             try {
                 Thread.sleep(12000);
             } catch (InterruptedException e) {
-                e.printStackTrace();
+                Thread.currentThread().interrupt();
             }
             threadIds = bean.findDeadlockedThreads();
         }
diff --git a/quickfixj-core/src/test/java/quickfix/mina/LostLogoutTest.java b/quickfixj-core/src/test/java/quickfix/mina/LostLogoutTest.java
@@ -144,7 +144,7 @@ public void fromApp(Message message, SessionID sessionId) throws FieldNotFound,
             try {
                 Thread.sleep(1000);
             } catch (InterruptedException e) {
-                e.printStackTrace();
+                Thread.currentThread().interrupt();
             }
             System.out.println("Server: message processing end");
         }
diff --git a/quickfixj-core/src/test/java/quickfix/mina/LostLogoutThreadedTest.java b/quickfixj-core/src/test/java/quickfix/mina/LostLogoutThreadedTest.java
@@ -144,7 +144,7 @@ public void fromApp(Message message, SessionID sessionId) throws FieldNotFound,
             try {
                 Thread.sleep(1000);
             } catch (InterruptedException e) {
-                e.printStackTrace();
+                Thread.currentThread().interrupt();
             }
             System.out.println("Server: message processing end");
         }
diff --git a/quickfixj-core/src/test/java/quickfix/mina/SingleThreadedEventHandlingStrategyTest.java b/quickfixj-core/src/test/java/quickfix/mina/SingleThreadedEventHandlingStrategyTest.java
@@ -68,7 +68,7 @@ public void cleanup() {
             try {
                 Thread.sleep(500);
             } catch (InterruptedException ex) {
-                // ignored
+                Thread.currentThread().interrupt();
             }
         }
     }
@@ -373,7 +373,7 @@ private static void assertQFJMessageProcessorThreads(int expected) {
             try {
                 Thread.sleep(100);
             } catch (InterruptedException ex) {
-                // ignored
+                Thread.currentThread().interrupt();
             }
             dumpAllThreads = bean.dumpAllThreads(false, false);
             qfjMPThreads = getMessageProcessorThreads(dumpAllThreads);
diff --git a/quickfixj-core/src/test/java/quickfix/mina/ssl/SSLAndNonSSLTest.java b/quickfixj-core/src/test/java/quickfix/mina/ssl/SSLAndNonSSLTest.java
@@ -214,6 +214,7 @@ public void run() {
                     shutdownLatch.await();
                 } catch (InterruptedException e1) {
                     try {
+                        Thread.currentThread().interrupt();
                         acceptor.stop(true);
                     } catch (RuntimeException e) {
                         e.printStackTrace();
diff --git a/quickfixj-core/src/test/java/quickfix/mina/ssl/SSLCertificateTest.java b/quickfixj-core/src/test/java/quickfix/mina/ssl/SSLCertificateTest.java
@@ -65,6 +65,7 @@ public void cleanup() {
         try {
             Thread.sleep(500);
         } catch (InterruptedException ex) {
+            Thread.currentThread().interrupt();
             java.util.logging.Logger.getLogger(SSLCertificateTest.class.getName()).log(Level.SEVERE, null, ex);
         }
     }
diff --git a/quickfixj-core/src/test/java/quickfix/test/acceptance/ConnectToServerStep.java b/quickfixj-core/src/test/java/quickfix/test/acceptance/ConnectToServerStep.java
@@ -61,7 +61,7 @@ public void run(TestResult result, TestConnection connection) {
             try {
                 Thread.sleep(reconnectDelay);
             } catch (InterruptedException e1) {
-                e1.printStackTrace();
+                Thread.currentThread().interrupt();
             }
         }
         try {
diff --git a/quickfixj-core/src/test/java/quickfix/test/acceptance/TestConnection.java b/quickfixj-core/src/test/java/quickfix/test/acceptance/TestConnection.java
@@ -68,8 +68,11 @@ private TestIoHandler getIoHandler(int clientId) {
 
     public void tearDown() {
         for (TestIoHandler testIoHandler : ioHandlers.values()) {
-            CloseFuture closeFuture = testIoHandler.getSession().closeNow();
-            closeFuture.awaitUninterruptibly();
+            IoSession session = testIoHandler.getSession();
+            if (session != null) {
+                CloseFuture closeFuture = session.closeNow();
+                closeFuture.awaitUninterruptibly();
+            }
         }
         ioHandlers.clear();
     }
@@ -165,6 +168,7 @@ public IoSession getSession() {
                     }
                 }
             } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
                 throw new RuntimeException(e);
             }
             return session;

Original file line number	Diff line number	Diff line change
`@@ -272,6 +272,7 @@ protected void waitForLogout() {`
`272`	`272`	`Thread.sleep(100L);`
`273`	`273`	`} catch (InterruptedException e) {`
`274`	`274`	`log.error(e.getMessage(), e);`
	`275`	`+ Thread.currentThread().interrupt();`
`275`	`276`	`}`
`276`	`277`	`final long elapsed = System.currentTimeMillis() - start;`
`277`	`278`	`Iterator<Session> sessionItr = loggedOnSessions.iterator();`
`@@ -375,6 +376,7 @@ public void run() {`
`375`	`376`	`try {`
`376`	`377`	`delegate.await();`
`377`	`378`	`} catch (InterruptedException e) {`
	`379`	`+ Thread.currentThread().interrupt();`
`378`	`380`	`}`
`379`	`381`	`}`
`380`	`382`
`@@ -449,9 +451,10 @@ public static void closeManagedSessionsAndDispose(IoService ioService, boolean a`
`449`	`451`	`completed = closeFuture.await(1000, TimeUnit.MILLISECONDS);`
`450`	`452`	`} catch (InterruptedException ex) {`
`451`	`453`	`Thread.currentThread().interrupt();`
`452`		`- }`
`453`		`- if (!completed) {`
`454`		`- logger.warn("Could not close IoSession {}", ioSession);`
	`454`	`+ } finally {`
	`455`	`+ if (!completed) {`
	`456`	`+ logger.warn("Could not close IoSession {}", ioSession);`
	`457`	`+ }`
`455`	`458`	`}`
`456`	`459`	`}`
`457`	`460`	`}`
Original file line number	Diff line number	Diff line change
`@@ -76,7 +76,7 @@ public void onMessage(Session quickfixSession, Message message) {`
`76`	`76`	`queueTracker.put(new SessionMessageEvent(quickfixSession, message));`
`77`	`77`	`} catch (InterruptedException e) {`
`78`	`78`	`isStopped = true;`
`79`		`- throw new RuntimeException(e);`
	`79`	`+ Thread.currentThread().interrupt();`
`80`	`80`	`}`
`81`	`81`	`}`
`82`	`82`
`@@ -196,7 +196,8 @@ public void stopHandlingMessages(boolean join) {`
`196`	`196`	`try {`
`197`	`197`	`messageProcessingThread.join();`
`198`	`198`	`} catch (InterruptedException e) {`
`199`		`- sessionConnector.log.error("{} interrupted.", MESSAGE_PROCESSOR_THREAD_NAME);`
	`199`	`+ sessionConnector.log.warn("{} interrupted.", MESSAGE_PROCESSOR_THREAD_NAME);`
	`200`	`+ Thread.currentThread().interrupt();`
`200`	`201`	`}`
`201`	`202`	`}`
`202`	`203`	`}`
Original file line number	Diff line number	Diff line change
`@@ -203,6 +203,7 @@ public void enqueue(Message message) {`
`203`	`203`	`queueTracker.put(message);`
`204`	`204`	`} catch (final InterruptedException e) {`
`205`	`205`	`quickfixSession.getLog().onErrorEvent(e.toString());`
	`206`	`+ Thread.currentThread().interrupt();`
`206`	`207`	`}`
`207`	`208`	`}`
`208`	`209`
`@@ -227,6 +228,7 @@ void doRun() {`
`227`	`228`	`LogUtil.logThrowable(quickfixSession.getSessionID(),`
`228`	`229`	`"Message dispatcher interrupted", e);`
`229`	`230`	`stopping = true;`
	`231`	`+ Thread.currentThread().interrupt();`
`230`	`232`	`} catch (final Throwable e) {`
`231`	`233`	`LogUtil.logThrowable(quickfixSession.getSessionID(),`
`232`	`234`	`"Error during message processing", e);`
Original file line number	Diff line number	Diff line change
`@@ -206,6 +206,7 @@ public void waitForLogon() {`
`206`	`206`	`try {`
`207`	`207`	`logonLatch.await(20, TimeUnit.SECONDS);`
`208`	`208`	`} catch (InterruptedException e) {`
	`209`	`+ Thread.currentThread().interrupt();`
`209`	`210`	`fail(e.getMessage());`
`210`	`211`	`}`
`211`	`212`	`}`
`@@ -220,6 +221,7 @@ public synchronized void waitForMessages() {`
`220`	`221`	`fail("Timed out waiting for message");`
`221`	`222`	`}`
`222`	`223`	`} catch (InterruptedException e) {`
	`224`	`+ Thread.currentThread().interrupt();`
`223`	`225`	`fail(e.getMessage());`
`224`	`226`	`}`
`225`	`227`	`}`
Original file line number	Diff line number	Diff line change
`@@ -134,6 +134,7 @@ public void waitForLogon() {`
`134`	`134`	`try {`
`135`	`135`	`logonLatch.await(10, TimeUnit.SECONDS);`
`136`	`136`	`} catch (InterruptedException e) {`
	`137`	`+ Thread.currentThread().interrupt();`
`137`	`138`	`fail(e.getMessage());`
`138`	`139`	`}`
`139`	`140`	`}`
`@@ -148,6 +149,7 @@ public synchronized void waitForMessages() {`
`148`	`149`	`fail("Timed out waiting for message");`
`149`	`150`	`}`
`150`	`151`	`} catch (InterruptedException e) {`
	`152`	`+ Thread.currentThread().interrupt();`
`151`	`153`	`fail(e.getMessage());`
`152`	`154`	`}`
`153`	`155`	`}`
`@@ -225,7 +227,7 @@ public void run() {`
`225`	`227`	`try {`
`226`	`228`	`Thread.sleep(12000);`
`227`	`229`	`} catch (InterruptedException e) {`
`228`		`- e.printStackTrace();`
	`230`	`+ Thread.currentThread().interrupt();`
`229`	`231`	`}`
`230`	`232`	`threadIds = bean.findDeadlockedThreads();`
`231`	`233`	`}`
Original file line number	Diff line number	Diff line change
`@@ -144,7 +144,7 @@ public void fromApp(Message message, SessionID sessionId) throws FieldNotFound,`
`144`	`144`	`try {`
`145`	`145`	`Thread.sleep(1000);`
`146`	`146`	`} catch (InterruptedException e) {`
`147`		`- e.printStackTrace();`
	`147`	`+ Thread.currentThread().interrupt();`
`148`	`148`	`}`
`149`	`149`	`System.out.println("Server: message processing end");`
`150`	`150`	`}`
Original file line number	Diff line number	Diff line change
`@@ -68,7 +68,7 @@ public void cleanup() {`
`68`	`68`	`try {`
`69`	`69`	`Thread.sleep(500);`
`70`	`70`	`} catch (InterruptedException ex) {`
`71`		`- // ignored`
	`71`	`+ Thread.currentThread().interrupt();`
`72`	`72`	`}`
`73`	`73`	`}`
`74`	`74`	`}`
`@@ -373,7 +373,7 @@ private static void assertQFJMessageProcessorThreads(int expected) {`
`373`	`373`	`try {`
`374`	`374`	`Thread.sleep(100);`
`375`	`375`	`} catch (InterruptedException ex) {`
`376`		`- // ignored`
	`376`	`+ Thread.currentThread().interrupt();`
`377`	`377`	`}`
`378`	`378`	`dumpAllThreads = bean.dumpAllThreads(false, false);`
`379`	`379`	`qfjMPThreads = getMessageProcessorThreads(dumpAllThreads);`
Original file line number	Diff line number	Diff line change
`@@ -65,6 +65,7 @@ public void cleanup() {`
`65`	`65`	`try {`
`66`	`66`	`Thread.sleep(500);`
`67`	`67`	`} catch (InterruptedException ex) {`
	`68`	`+ Thread.currentThread().interrupt();`
`68`	`69`	`java.util.logging.Logger.getLogger(SSLCertificateTest.class.getName()).log(Level.SEVERE, null, ex);`
`69`	`70`	`}`
`70`	`71`	`}`