NEU

[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 230f822..bff38ae 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
@@ -68,6 +68,50 @@ public class ChatMessageChannel implements Runnable, ConsumerRebalanceListener
   }
 
 
+  Mono<Message> sendMessage(
+      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, MessageTo> record =
+          new ProducerRecord<>(
+              tp.topic(),
+              tp.partition(),
+              zdt.toEpochSecond(),
+              chatRoomId.toString(),
+              MessageTo.of(key.getUsername(), key.getMessageId(), text));
+
+      producer.send(record, ((metadata, exception) ->
+      {
+        if (metadata != null)
+        {
+          // On successful send
+          Message message = new Message(key, metadata.offset(), timestamp, text);
+          log.info("Successfully send message {}", message);
+          sink.success(message);
+        }
+        else
+        {
+          // On send-failure
+          log.error(
+              "Could not send message for chat-room={}, key={}, timestamp={}, text={}: {}",
+              chatRoomId,
+              key,
+              timestamp,
+              text,
+              exception);
+          sink.error(exception);
+        }
+      }));
+    });
+  }
+
   @Override
   public void onPartitionsAssigned(Collection<TopicPartition> partitions)
   {
@@ -147,9 +191,8 @@ public class ChatMessageChannel implements Runnable, ConsumerRebalanceListener
       }
       catch (WakeupException e)
       {
-      }
-      catch (RecordDeserializationException e)
-      {
+        log.info("Received WakeupException, exiting!");
+        running = false;
       }
     }
   }
@@ -202,84 +245,24 @@ public class ChatMessageChannel implements Runnable, ConsumerRebalanceListener
         .toList());
   }
 
-  Mono<Message> sendMessage(
-      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, MessageTo> record =
-          new ProducerRecord<>(
-              tp.topic(),
-              tp.partition(),
-              zdt.toEpochSecond(),
-              chatRoomId.toString(),
-              MessageTo.of(key.getUsername(), key.getMessageId(), text));
-
-      producer.send(record, ((metadata, exception) ->
-      {
-        if (metadata != null)
-        {
-          // On successful send
-          Message message = new Message(key, metadata.offset(), timestamp, text);
-          log.info("Successfully send message {}", message);
-          sink.success(message);
-        }
-        else
-        {
-          // On send-failure
-          log.error(
-              "Could not send message for chat-room={}, key={}, timestamp={}, text={}: {}",
-              chatRoomId,
-              key,
-              timestamp,
-              text,
-              exception);
-          sink.error(exception);
-        }
-      }));
-    });
-  }
-
 
   void putChatRoom(ChatRoom chatRoom)
   {
     Integer partition = chatRoom.getShard();
     UUID chatRoomId = chatRoom.getId();
-    ChatRoom existingChatRoom = chatrooms[partition].get(chatRoomId);
-    if (existingChatRoom == null)
+    if (chatrooms[partition].containsKey(chatRoomId))
+    {
+      log.warn("Ignoring existing chat-room: " + chatRoom);
+    }
+    else
     {
       log.info(
-          "Creating new chat-room in partition {}: {}",
+          "Adding new chat-room to partition {}: {}",
           partition,
           chatRoom);
+
       chatrooms[partition].put(chatRoomId, chatRoom);
     }
-    else
-    {
-      if (chatRoom.getShard() != existingChatRoom.getShard())
-      {
-        throw new IllegalArgumentException(
-            "Could not change the shard of existing chat-room " +
-            chatRoomId + " from " +
-            existingChatRoom.getShard() + " to " +
-            chatRoom.getShard());
-      }
-      else
-      {
-        log.info(
-            "Updating chat-room in partition {}: {} -> {}",
-            partition,
-            existingChatRoom,
-            chatRoom);
-        existingChatRoom.s
-      }
-    }
   }
 
   Mono<ChatRoom> getChatRoom(int shard, UUID id)