NG

[demos/kafka/chat] / src / main / java / de / juplo / kafka / chat / backend / persistence / kafka / ChatMessageChannel.java

diff --git a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatMessageChannel.java b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatMessageChannel.java
index 8294316..94f6fa6 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatMessageChannel.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/ChatMessageChannel.java
@@ -1,8 +1,8 @@
 package de.juplo.kafka.chat.backend.persistence.kafka;
 
 import de.juplo.kafka.chat.backend.domain.ChatRoom;
+import de.juplo.kafka.chat.backend.domain.ChatRoomInfo;
 import de.juplo.kafka.chat.backend.domain.Message;
-import de.juplo.kafka.chat.backend.persistence.KafkaLikeShardingStrategy;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.Consumer;
@@ -29,11 +29,12 @@ public class ChatMessageChannel implements Runnable, ConsumerRebalanceListener
   private final Consumer<String, AbstractTo> consumer;
   private final ZoneId zoneId;
   private final int numShards;
+  private final int bufferSize;
+  private final Clock clock;
   private final boolean[] isShardOwned;
   private final long[] currentOffset;
   private final long[] nextOffset;
   private final Map<UUID, ChatRoom>[] chatrooms;
-  private final KafkaLikeShardingStrategy shardingStrategy;
 
   private boolean running;
   @Getter
@@ -45,7 +46,9 @@ public class ChatMessageChannel implements Runnable, ConsumerRebalanceListener
     Producer<String, AbstractTo> producer,
     Consumer<String, AbstractTo> consumer,
     ZoneId zoneId,
-    int numShards)
+    int numShards,
+    int bufferSize,
+    Clock clock)
   {
     log.debug(
         "Creating ChatMessageChannel for topic {} with {} partitions",
@@ -56,6 +59,8 @@ public class ChatMessageChannel implements Runnable, ConsumerRebalanceListener
     this.producer = producer;
     this.zoneId = zoneId;
     this.numShards = numShards;
+    this.bufferSize = bufferSize;
+    this.clock = clock;
     this.isShardOwned = new boolean[numShards];
     this.currentOffset = new long[numShards];
     this.nextOffset = new long[numShards];
@@ -63,25 +68,59 @@ public class ChatMessageChannel implements Runnable, ConsumerRebalanceListener
     IntStream
         .range(0, numShards)
         .forEach(shard -> this.chatrooms[shard] = new HashMap<>());
-    this.shardingStrategy = new KafkaLikeShardingStrategy(numShards);
   }
 
 
-  Mono<Message> sendMessage(
+
+  Mono<ChatRoomInfo> sendCreateChatRoomRequest(
+      UUID chatRoomId,
+      String name)
+  {
+    CreateChatRoomRequestTo createChatRoomRequestTo = CreateChatRoomRequestTo.of(name);
+    return Mono.create(sink ->
+    {
+      ProducerRecord<String, CreateChatRoomRequestTo> record =
+          new ProducerRecord<>(
+              topic,
+              chatRoomId.toString(),
+              createChatRoomRequestTo);
+
+      producer.send(record, ((metadata, exception) ->
+      {
+        if (metadata != null)
+        {
+          log.info("Successfully send chreate-request for chat room: {}", createChatRoomRequestTo);
+          ChatRoomInfo chatRoomInfo = ChatRoomInfo.of(chatRoomId, name, record.partition());
+          createChatRoom(chatRoomInfo);
+          sink.success(chatRoomInfo);
+        }
+        else
+        {
+          // On send-failure
+          log.error(
+              "Could not send create-request for chat room (id={}, name={}): {}",
+              chatRoomId,
+              name,
+              exception);
+          sink.error(exception);
+        }
+      }));
+    });
+  }
+
+  Mono<Message> sendChatMessage(
       UUID chatRoomId,
       Message.MessageKey key,
       LocalDateTime timestamp,
       String text)
   {
-    int shard = this.shardingStrategy.selectShard(chatRoomId);
-    TopicPartition tp = new TopicPartition(topic, shard);
     ZonedDateTime zdt = ZonedDateTime.of(timestamp, zoneId);
     return Mono.create(sink ->
     {
-      ProducerRecord<String, ChatMessageTo> record =
+      ProducerRecord<String, AbstractTo> record =
           new ProducerRecord<>(
-              tp.topic(),
-              tp.partition(),
+              topic,
+              null,
               zdt.toEpochSecond(),
               chatRoomId.toString(),
               ChatMessageTo.of(key.getUsername(), key.getMessageId(), text));
@@ -202,14 +241,18 @@ public class ChatMessageChannel implements Runnable, ConsumerRebalanceListener
   {
     for (ConsumerRecord<String, AbstractTo> record : records)
     {
+      UUID chatRoomId = UUID.fromString(record.key());
+
       switch (record.value().getType())
       {
         case CREATE_CHATROOM_REQUEST:
-          createChatRoom((CreateChatRoomRequestTo) record.value());
+          createChatRoom(
+              chatRoomId,
+              (CreateChatRoomRequestTo) record.value(),
+              record.partition());
           break;
 
         case MESSAGE_SENT:
-          UUID chatRoomId = UUID.fromString(record.key());
           Instant instant = Instant.ofEpochSecond(record.timestamp());
           LocalDateTime timestamp = LocalDateTime.ofInstant(instant, zoneId);
           loadChatMessage(
@@ -219,6 +262,13 @@ public class ChatMessageChannel implements Runnable, ConsumerRebalanceListener
               (ChatMessageTo) record.value(),
               record.partition());
           break;
+
+        default:
+          log.debug(
+              "Ignoring message for chat-room {} with offset {}: {}",
+              chatRoomId,
+              record.offset(),
+              record.value());
       }
 
       nextOffset[record.partition()] = record.offset() + 1;
@@ -226,10 +276,26 @@ public class ChatMessageChannel implements Runnable, ConsumerRebalanceListener
   }
 
   void createChatRoom(
+      UUID chatRoomId,
       CreateChatRoomRequestTo createChatRoomRequestTo,
       int partition)
   {
-    chatrooms[partition].put
+    putChatRoom(ChatRoomInfo.of(
+        chatRoomId,
+        createChatRoomRequestTo.getName(),
+        partition));
+  }
+
+
+  void createChatRoom(ChatRoomInfo chatRoomInfo)
+  {
+    UUID id = chatRoomInfo.getId();
+    String name = chatRoomInfo.getName();
+    int shard = chatRoomInfo.getShard();
+    log.info("Creating ChatRoom {} with buffer-size {}", id, bufferSize);
+    KafkaChatRoomService service = new KafkaChatRoomService(this, id);
+    ChatRoom chatRoom = new ChatRoom(id, name, shard, clock, service, bufferSize);
+    putChatRoom(chatRoom);
   }
 
   void loadChatMessage(
@@ -267,7 +333,7 @@ public class ChatMessageChannel implements Runnable, ConsumerRebalanceListener
   }
 
 
-  void putChatRoom(ChatRoom chatRoom)
+  private void putChatRoom(ChatRoom chatRoom)
   {
     Integer partition = chatRoom.getShard();
     UUID chatRoomId = chatRoom.getId();
@@ -290,13 +356,4 @@ public class ChatMessageChannel implements Runnable, ConsumerRebalanceListener
   {
     return Mono.justOrEmpty(chatrooms[shard].get(id));
   }
-
-  Flux<ChatRoom> getChatRooms()
-  {
-    return Flux.fromStream(IntStream
-        .range(0, numShards)
-        .filter(shard -> isShardOwned[shard])
-        .mapToObj(shard -> Integer.valueOf(shard))
-        .flatMap(shard -> chatrooms[shard].values().stream()));
-  }
 }