WIP

diff --git a/src/main/java/de/juplo/kafka/chat/backend/domain/ChatRoomInfo.java b/src/main/java/de/juplo/kafka/chat/backend/domain/ChatRoomInfo.java
index 33c522d..e91b28c 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/domain/ChatRoomInfo.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/domain/ChatRoomInfo.java
@@ -19,4 +19,10 @@ public class ChatRoomInfo
   private final String name;
   @Getter
   private final Integer shard;
+
+
+  public ChatRoomInfo(UUID id, String name)
+  {
+    this(id, name, null);
+  }
 }

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 22c4668..5a1d186 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
@@ -30,6 +30,7 @@ public class InfoChannel implements Runnable
   private final long[] startOffset;
   private final long[] currentOffset;
   private final Map<UUID, ChatRoomInfo> chatRoomInfo;
+  private final DataChannel dataChannel;
 
   private boolean running;
 
@@ -37,7 +38,8 @@ public class InfoChannel implements Runnable
   public InfoChannel(
     String topic,
     Producer<String, AbstractMessageTo> producer,
-    Consumer<String, AbstractMessageTo> infoChannelConsumer)
+    Consumer<String, AbstractMessageTo> infoChannelConsumer,
+    DataChannel dataChannel)
   {
     log.debug(
         "Creating InfoChannel for topic {}",
@@ -60,15 +62,16 @@ public class InfoChannel implements Runnable
         .entrySet()
         .stream()
         .forEach(entry -> this.startOffset[entry.getKey().partition()] = entry.getValue());
+
+    this.dataChannel = dataChannel;
   }
 
 
-  Mono<Boolean> loadInProgress()
+  boolean loadInProgress()
   {
-    return Mono
-        .fromSupplier(() -> IntStream
-            .range(0, numShards)
-            .anyMatch(partition -> currentOffset[partition] < startOffset[partition]));
+    return IntStream
+        .range(0, numShards)
+        .anyMatch(partition -> currentOffset[partition] < startOffset[partition]);
   }
 
   Mono<ChatRoomInfo> sendChatRoomCreatedEvent(
@@ -176,6 +179,7 @@ public class InfoChannel implements Runnable
           chatRoomId);
 
       this.chatRoomInfo.put(chatRoomId, chatRoomInfo);
+      this.dataChannel.createChatRoom(chatRoomInfo);
     }
   }
 

diff --git a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaChatHomeService.java b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaChatHomeService.java
index d0b8eff..9832519 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaChatHomeService.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaChatHomeService.java
@@ -40,10 +40,7 @@ public class KafkaChatHomeService implements ChatHomeService
   {
     return infoChannel
         .getChatRoomInfo(id)
-        .switchIfEmpty(Mono.error(() -> new UnknownChatroomException(
-            id,
-            shard,
-            dataChannel.getOwnedShards())));
+        .switchIfEmpty(Mono.error(() -> new UnknownChatroomException(id)));
   }
 
   @Override

diff --git a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesApplicationRunner.java b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesApplicationRunner.java
index a35cee0..125227a 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesApplicationRunner.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesApplicationRunner.java
@@ -6,6 +6,7 @@ import jakarta.annotation.PreDestroy;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.common.TopicPartition;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.ApplicationArguments;
 import org.springframework.boot.ApplicationRunner;
@@ -30,19 +31,44 @@ public class KafkaServicesApplicationRunner implements ApplicationRunner
   private final Consumer<String, AbstractMessageTo> chatRoomChannelConsumer;
   private final WorkAssignor workAssignor;
 
-  CompletableFuture<Void> chatRoomChannelConsumerJob;
+  CompletableFuture<Void> infoChannelConsumerJob;
+  CompletableFuture<Void> dataChannelConsumerJob;
 
 
   @Override
   public void run(ApplicationArguments args) throws Exception
   {
+    String infoTopic = properties.getKafka().getInfoChannelTopic();
+    List<TopicPartition> partitions = infoChannelConsumer
+        .partitionsFor(infoTopic)
+        .stream()
+        .map(partitionInfo -> new TopicPartition(
+            infoTopic,
+            partitionInfo.partition()))
+        .toList();
+    infoChannelConsumer.assign(partitions);
+    log.info("Starting the consumer for the InfoChannel");
+    infoChannelConsumerJob = taskExecutor
+        .submitCompletable(infoChannel)
+        .exceptionally(e ->
+        {
+          log.error("The consumer for the InfoChannel exited abnormally!", e);
+          return null;
+        });
+
+    while (infoChannel.loadInProgress())
+    {
+      log.info("InfoChannel is still loading...");
+      Thread.sleep(1000);
+    }
+
     workAssignor.assignWork(chatRoomChannelConsumer);
-    log.info("Starting the consumer for the ChatRoomChannel");
-    chatRoomChannelConsumerJob = taskExecutor
+    log.info("Starting the consumer for the DataChannel");
+    dataChannelConsumerJob = taskExecutor
         .submitCompletable(dataChannel)
         .exceptionally(e ->
         {
-          log.error("The consumer for the ChatRoomChannel exited abnormally!", e);
+          log.error("The consumer for the DataChannel exited abnormally!", e);
           return null;
         });
   }
@@ -51,9 +77,9 @@ public class KafkaServicesApplicationRunner implements ApplicationRunner
   public void joinChatRoomChannelConsumerJob()
   {
     log.info("Signaling the consumer of the CahtRoomChannel to quit its work");
-    chatRoomChannelConsumer.wakeup();
+    infoChannelConsumer.wakeup();
     log.info("Waiting for the consumer of the ChatRoomChannel to finish its work");
-    chatRoomChannelConsumerJob.join();
+    dataChannelConsumerJob.join();
     log.info("Joined the consumer of the ChatRoomChannel");
   }
 

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 ed1ccdd..018dc65 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
@@ -3,7 +3,6 @@ package de.juplo.kafka.chat.backend.implementation.kafka;
 import de.juplo.kafka.chat.backend.ChatBackendProperties;
 import de.juplo.kafka.chat.backend.domain.ChatHomeService;
 import de.juplo.kafka.chat.backend.implementation.kafka.messages.AbstractMessageTo;
-import de.juplo.kafka.chat.backend.implementation.kafka.messages.data.CommandCreateChatRoomTo;
 import de.juplo.kafka.chat.backend.implementation.kafka.messages.data.EventChatMessageReceivedTo;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;

diff --git a/src/main/java/de/juplo/kafka/chat/backend/storage/mongodb/MongoDbStorageStrategy.java b/src/main/java/de/juplo/kafka/chat/backend/storage/mongodb/MongoDbStorageStrategy.java
index c87036c..3eb9096 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/storage/mongodb/MongoDbStorageStrategy.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/storage/mongodb/MongoDbStorageStrategy.java
@@ -36,18 +36,7 @@ public class MongoDbStorageStrategy implements StorageStrategy
         .map(chatRoomTo ->
         {
           UUID chatRoomId = UUID.fromString(chatRoomTo.getId());
-          int shard = shardingStrategy.selectShard(chatRoomId);
-
-          log.info(
-              "{} - old shard: {}, new shard:  {}",
-              chatRoomId,
-              chatRoomTo.getShard(),
-              shard);
-
-          return new ChatRoomInfo(
-              chatRoomId,
-              chatRoomTo.getName(),
-              shard);
+          return new ChatRoomInfo(chatRoomId, chatRoomTo.getName());
         });
   }