WIP:refactor: Refined channel-states, introduced `ChannelState` -- ALIGN

diff --git a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/ChannelTaskExecutor.java b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/ChannelTaskExecutor.java
index e61ecfe..d274cf2 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/ChannelTaskExecutor.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/ChannelTaskExecutor.java
@@ -38,7 +38,7 @@ public class ChannelTaskExecutor
   }
 
   @PreDestroy
-  public void joinConsumerTaskJob()
+  public void join()
   {
     log.info("Signaling the consumer-task for {} to quit its work", channel);
     consumer.wakeup();

diff --git a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/ChannelTaskRunner.java b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/ChannelTaskRunner.java
index 015f7d5..9e8576d 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/ChannelTaskRunner.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/ChannelTaskRunner.java
@@ -19,14 +19,20 @@ public class ChannelTaskRunner
 
   public void joinChannel() throws InterruptedException
   {
-    dataChannelTaskExecutor.joinConsumerTaskJob();
-    while (infoChannel.getChannelState() != ChannelState.SHUTTING_DOWN)
+    joinChannel(dataChannelTaskExecutor);
+    joinChannel(infoChannelTaskExecutor);
+  }
+
+  private void joinChannel(
+      ChannelTaskExecutor channelTaskExecutor)
+      throws InterruptedException
+  {
+    Channel channel = channelTaskExecutor.getChannel();
+    while (channel.getChannelState() != ChannelState.SHUTTING_DOWN)
     {
-      log.info("Waiting for {} to shut down...", infoChannel);
+      log.info("Waiting for {} to shut down...", channel);
       Thread.sleep(1000);
     }
-    infoChannelTaskExecutor.joinConsumerTaskJob();
+    channelTaskExecutor.join();
   }
-
-  private void joinChannel()
 }

diff --git a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/DataChannel.java b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/DataChannel.java
index 0b572f4..2468af5 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/DataChannel.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/DataChannel.java
@@ -230,6 +230,7 @@ public class DataChannel implements Channel, ConsumerRebalanceListener
       catch (WakeupException e)
       {
         log.info("Received WakeupException, exiting!");
+        channelState = ChannelState.SHUTTING_DOWN;
         running = false;
       }
     }

diff --git a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/InfoChannel.java b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/InfoChannel.java
index ec4a5eb..7665fae 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/InfoChannel.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/InfoChannel.java
@@ -210,6 +210,7 @@ public class InfoChannel implements Channel
       catch (WakeupException e)
       {
         log.info("Received WakeupException, exiting!");
+        channelState = ChannelState.SHUTTING_DOWN;
         running = false;
       }
     }

diff --git a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesConfiguration.java b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesConfiguration.java
index 1088cab..857f8ec 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesConfiguration.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesConfiguration.java
@@ -46,8 +46,7 @@ public class KafkaServicesConfiguration
   {
     return new ChannelTaskRunner(
         infoChannelTaskExecutor,
-        dataChannelTaskExecutor,
-        infoChannel);
+        dataChannelTaskExecutor);
   }
 
   @Bean