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..8a53d3c 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,6 +1,7 @@
 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;
@@ -67,7 +68,44 @@ public class ChatMessageChannel implements Runnable, ConsumerRebalanceListener
   }
 
 
-  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,
@@ -202,14 +240,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(
@@ -226,10 +268,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 +325,7 @@ public class ChatMessageChannel implements Runnable, ConsumerRebalanceListener
   }
 
 
-  void putChatRoom(ChatRoom chatRoom)
+  private void putChatRoom(ChatRoom chatRoom)
   {
     Integer partition = chatRoom.getShard();
     UUID chatRoomId = chatRoom.getId();