Fix thread interrupt handling

csowada · Jan 1, 2020 · 9bb7d72 · 9bb7d72
1 parent 2ac893c
commit 9bb7d72
Show file tree

Hide file tree

Showing 3 changed files with 109 additions and 37 deletions.
diff --git a/src/main/java/de/csdev/ebus/core/EBusEbusdController.java b/src/main/java/de/csdev/ebus/core/EBusEbusdController.java
@@ -11,6 +11,7 @@
 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStreamReader;
+import java.io.InterruptedIOException;
 import java.io.OutputStreamWriter;
 import java.io.Writer;
 import java.net.Socket;
@@ -116,7 +117,7 @@ public void run() {
         }
     }
 
-    private ByteBuffer parseLine(String readLine) throws IOException {
+    private ByteBuffer parseLine(String readLine) throws IOException, InterruptedException {
 
         ByteBuffer b = null;
 
@@ -239,6 +240,14 @@ public void run() {
 
                     }
 
+                } catch (InterruptedIOException e) {
+                    // re-enable the interrupt to stop the while loop
+                    Thread.currentThread().interrupt();
+
+                } catch (InterruptedException e) {
+                    // re-enable the interrupt to stop the while loop
+                    Thread.currentThread().interrupt();
+
                 } catch (IOException e) {
                     logger.error("error!", e);
                     fireOnConnectionException(e);
@@ -252,9 +261,7 @@ public void run() {
                 }
 
             } // while loop
-        } catch (
-
-        Exception e) {
+        } catch (Exception e) {
             logger.error("error!", e);
         }
 
@@ -295,7 +302,7 @@ public ByteBuffer convertEBusdDataToFullTelegram(byte[] masterData, byte[] slave
         return b;
     }
 
-    private void reconnect() throws IOException {
+    private void reconnect() throws IOException, InterruptedException {
 
         if (!isRunning()) {
             logger.trace("Skip reconnect, thread was interrupted ...");
@@ -316,17 +323,12 @@ private void reconnect() throws IOException {
             reConnectCounter++;
 
             logger.warn("Retry to connect to ebusd daemon in {} seconds ...", 5 * reConnectCounter);
-            try {
 
-                Thread.sleep(5000 * reConnectCounter);
-                disconnect();
+            Thread.sleep(5000 * reConnectCounter);
+            disconnect();
 
-                if (connect()) {
-                    resetWatchdogTimer();
-                }
-
-            } catch (InterruptedException e) {
-                // noop, accept interruptions
+            if (connect()) {
+                resetWatchdogTimer();
             }
         }
 

diff --git a/src/main/java/de/csdev/ebus/core/EBusLowLevelController.java b/src/main/java/de/csdev/ebus/core/EBusLowLevelController.java
@@ -9,6 +9,7 @@
 package de.csdev.ebus.core;
 
 import java.io.IOException;
+import java.io.InterruptedIOException;
 import java.nio.BufferOverflowException;
 
 import org.slf4j.Logger;
@@ -102,7 +103,7 @@ private void onEBusDataReceived(byte data) throws IOException {
 
     }
 
-    private void reconnect() throws IOException {
+    private void reconnect() throws IOException, InterruptedException {
 
         if (!isRunning()) {
             logger.trace("Skip reconnect, thread was interrupted ...");
@@ -123,16 +124,12 @@ private void reconnect() throws IOException {
             reConnectCounter++;
 
             logger.warn("Retry to connect to eBUS adapter in {} seconds ...", 5 * reConnectCounter);
-            try {
-                Thread.sleep(5000 * reConnectCounter);
 
-                connection.close();
-                if (connection.open()) {
-                    resetWatchdogTimer();
-                }
+            Thread.sleep(5000 * reConnectCounter);
 
-            } catch (InterruptedException e) {
-                // noop, accept interruptions
+            connection.close();
+            if (connection.open()) {
+                resetWatchdogTimer();
             }
         }
 
@@ -182,7 +179,10 @@ public void run() {
         resetWatchdogTimer();
 
         // loop until interrupt or reconnector count is -1 (to many retries)
-        while (!(isInterrupted() || reConnectCounter == -1)) {
+
+        // while (!Thread.currentThread().isInterrupted() || reConnectCounter != -1) {
+
+        while (!(this.isInterrupted() || reConnectCounter == -1)) {
             try {
 
                 if (!connection.isOpen()) {
@@ -212,6 +212,11 @@ public void run() {
                     }
                 }
 
+            } catch (InterruptedIOException e) {
+                Thread.currentThread().interrupt();
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+
             } catch (IOException e) {
                 logger.error("An IO exception has occured! Try to reconnect eBUS connector ...", e);
                 fireOnConnectionException(e);
@@ -220,6 +225,8 @@ public void run() {
                     reconnect();
                 } catch (IOException e1) {
                     logger.error(e.toString(), e1);
+                } catch (InterruptedException e1) {
+                    Thread.currentThread().interrupt();
                 }
 
             } catch (BufferOverflowException e) {
@@ -410,7 +417,7 @@ else if ((byte) (readByte & 0x0F) == (byte) (b & 0x0F)) {
     @Override
     protected void dispose() {
 
-        logger.debug("eBUS connection thread is shuting down ...");
+        logger.info("eBUS connection thread is shuting down ...");
 
         super.dispose();
 
@@ -443,9 +450,4 @@ protected void fireWatchDogTimer() {
         }
     }
 
-    // @Override
-    // public ConnectionStatus getConnectionStatus() {
-    // return this.connectionStatus;
-    // }
-
 }
diff --git a/src/test/java/de/csdev/ebus/core/EBusControllerTest.java b/src/test/java/de/csdev/ebus/core/EBusControllerTest.java
@@ -9,13 +9,12 @@
 package de.csdev.ebus.core;
 
 import java.io.IOException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 
 import org.junit.Assert;
 import org.junit.Test;
 
 import de.csdev.ebus.core.connection.EBusEmulatorConnection;
+import de.csdev.ebus.core.connection.EBusTCPConnection;
 
 public class EBusControllerTest {
 
@@ -27,17 +26,86 @@ public void testWatchdogTimeout() throws InterruptedException, IOException, EBus
 
         controller.setWatchdogTimerTimeout(1);
 
-        ExecutorService threadExecutor = Executors
-                .newSingleThreadExecutor(new EBusWorkerThreadFactory("ebus-controller", false));
-        threadExecutor.execute(controller);
+        controller.start();
+
+        Thread.sleep(50);
+        Assert.assertTrue(controller.getConnection().isOpen());
 
         Thread.sleep(500);
-        controller.start();
         Assert.assertTrue(controller.getConnection().isOpen());
 
+        Thread.sleep(1000);
+        Assert.assertFalse(controller.getConnection().isOpen());
+    }
+
+    @Test
+    public void testInterruptEmulator() throws InterruptedException, IOException, EBusControllerException {
+
+        EBusEmulatorConnection connection = new EBusEmulatorConnection(false);
+        EBusLowLevelController controller = new EBusLowLevelController(connection);
+
+        controller.setWatchdogTimerTimeout(10);
+
+        Thread.sleep(500);
+        controller.start();
+
         Thread.sleep(1500);
 
-        Assert.assertFalse(controller.getConnection().isOpen());
+        Assert.assertTrue(controller.isAlive());
+        Assert.assertFalse(controller.isInterrupted());
+
+        controller.interrupt();
+
+        Thread.sleep(500);
+
+        Assert.assertFalse(controller.isAlive());
+        Assert.assertFalse(controller.isInterrupted());
+    }
+
+    @Test
+    public void testInterruptTCPRaw() throws InterruptedException, IOException, EBusControllerException {
+
+        EBusTCPConnection connection = new EBusTCPConnection("localhost", 1);
+        EBusLowLevelController controller = new EBusLowLevelController(connection);
+
+        controller.setWatchdogTimerTimeout(10);
+
+        Thread.sleep(500);
+        controller.start();
+
+        Thread.sleep(1500);
+
+        Assert.assertTrue(controller.isAlive());
+        Assert.assertFalse(controller.isInterrupted());
+
+        controller.interrupt();
+
+        Thread.sleep(2500);
+
+        Assert.assertFalse(controller.isAlive());
+        Assert.assertFalse(controller.isInterrupted());
     }
 
+    @Test
+    public void testInterruptEbusd() throws InterruptedException, IOException, EBusControllerException {
+
+        EBusEbusdController controller = new EBusEbusdController("localhost", 1);
+
+        controller.setWatchdogTimerTimeout(10);
+
+        Thread.sleep(500);
+        controller.start();
+
+        Thread.sleep(1500);
+
+        Assert.assertTrue(controller.isAlive());
+        Assert.assertFalse(controller.isInterrupted());
+
+        controller.interrupt();
+
+        Thread.sleep(500);
+
+        Assert.assertFalse(controller.isAlive());
+        Assert.assertFalse(controller.isInterrupted());
+    }
 }