From: Kai Moritz <kai@juplo.de>
Date: Mon, 4 Mar 2024 14:33:11 +0000 (+0100)
Subject: WIP:refactor: Refined channel-states, introduced `ChannelState` -- ALIGN
X-Git-Tag: rebase--2024-03-05--09-07~3
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=fe12b154e9817bbefe01f3c6b8173f08525f6474;p=demos%2Fkafka%2Fchat

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 e61ecfe7..d274cf24 100644
--- 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 015f7d51..9e8576d0 100644
--- 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 0b572f48..2468af5a 100644
--- 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 ec4a5ebd..7665faea 100644
--- 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 1088cabf..857f8ecd 100644
--- 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