feat: Introduced interface `ShardingPublisherStrategy`

[demos/kafka/chat] / src / main / java / de / juplo / kafka / chat / backend / implementation / kafka / KafkaServicesConfiguration.java

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 5bde07c..c3027fa 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
@@ -2,6 +2,7 @@ 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.domain.ShardingPublisherStrategy;
 import de.juplo.kafka.chat.backend.implementation.kafka.messages.AbstractMessageTo;
 import de.juplo.kafka.chat.backend.implementation.kafka.messages.data.EventChatMessageReceivedTo;
 import de.juplo.kafka.chat.backend.implementation.kafka.messages.info.EventChatRoomCreated;
@@ -20,6 +21,7 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.support.serializer.JsonDeserializer;
 import org.springframework.kafka.support.serializer.JsonSerializer;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import reactor.core.publisher.Mono;
 
 import java.time.Clock;
 import java.time.ZoneId;
@@ -39,11 +41,13 @@ public class KafkaServicesConfiguration
   @Bean
   ConsumerTaskRunner consumerTaskRunner(
       ConsumerTaskExecutor infoChannelConsumerTaskExecutor,
-      ConsumerTaskExecutor dataChannelConsumerTaskExecutor)
+      ConsumerTaskExecutor dataChannelConsumerTaskExecutor,
+      InfoChannel infoChannel)
   {
     return new ConsumerTaskRunner(
         infoChannelConsumerTaskExecutor,
-        dataChannelConsumerTaskExecutor);
+        dataChannelConsumerTaskExecutor,
+        infoChannel);
   }
 
   @Bean
@@ -51,7 +55,7 @@ public class KafkaServicesConfiguration
       ThreadPoolTaskExecutor taskExecutor,
       InfoChannel infoChannel,
       Consumer<String, AbstractMessageTo> infoChannelConsumer,
-      ConsumerTaskExecutor.WorkAssignor infoChannelWorkAssignor)
+      WorkAssignor infoChannelWorkAssignor)
   {
     return new ConsumerTaskExecutor(
         taskExecutor,
@@ -61,8 +65,7 @@ public class KafkaServicesConfiguration
   }
 
   @Bean
-  ConsumerTaskExecutor.WorkAssignor infoChannelWorkAssignor(
-      ChatBackendProperties properties)
+  WorkAssignor infoChannelWorkAssignor(ChatBackendProperties properties)
   {
     return consumer ->
     {
@@ -82,7 +85,7 @@ public class KafkaServicesConfiguration
       ThreadPoolTaskExecutor taskExecutor,
       DataChannel dataChannel,
       Consumer<String, AbstractMessageTo> dataChannelConsumer,
-      ConsumerTaskExecutor.WorkAssignor dataChannelWorkAssignor)
+      WorkAssignor dataChannelWorkAssignor)
   {
     return new ConsumerTaskExecutor(
         taskExecutor,
@@ -92,7 +95,7 @@ public class KafkaServicesConfiguration
   }
 
   @Bean
-  ConsumerTaskExecutor.WorkAssignor dataChannelWorkAssignor(
+  WorkAssignor dataChannelWorkAssignor(
       ChatBackendProperties properties,
       DataChannel dataChannel)
   {
@@ -125,7 +128,8 @@ public class KafkaServicesConfiguration
     return new InfoChannel(
         properties.getKafka().getInfoChannelTopic(),
         producer,
-        infoChannelConsumer);
+        infoChannelConsumer,
+        properties.getKafka().getInstanceUri());
   }
 
   @Bean
@@ -134,16 +138,21 @@ public class KafkaServicesConfiguration
       Producer<String, AbstractMessageTo> producer,
       Consumer<String, AbstractMessageTo> dataChannelConsumer,
       ZoneId zoneId,
-      Clock clock)
+      Clock clock,
+      InfoChannel infoChannel,
+      ShardingPublisherStrategy shardingPublisherStrategy)
   {
     return new DataChannel(
+        properties.getInstanceId(),
         properties.getKafka().getDataChannelTopic(),
         producer,
         dataChannelConsumer,
         zoneId,
         properties.getKafka().getNumPartitions(),
         properties.getChatroomBufferSize(),
-        clock);
+        clock,
+        infoChannel,
+        shardingPublisherStrategy);
   }
 
   @Bean
@@ -275,6 +284,18 @@ public class KafkaServicesConfiguration
     return properties;
   }
 
+  @Bean
+  ShardingPublisherStrategy shardingPublisherStrategy()
+  {
+    return new ShardingPublisherStrategy() {
+      @Override
+      public Mono<String> publishOwnership(int shard)
+      {
+        return Mono.just(Integer.toString(shard));
+      }
+    };
+  }
+
   @Bean
   ZoneId zoneId()
   {