WIP

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

diff --git a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatHomeService.java b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatHomeService.java
index a5d63fd..e171bc5 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatHomeService.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatHomeService.java
@@ -18,7 +18,8 @@ public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceL
   private final Consumer<String, MessageTo> consumer;
   private final String topic;
   // private final long[] offsets; Erst mal immer alles neu einlesen
-  private final Map<UUID, ChatRoom>[] chatrooms;
+  private final Map<UUID, KafkaChatRoomService>[] kafkaChatRoomServiceMaps;
+  private final Map<UUID, ChatRoom>[] chatRoomMaps;
 
 
   public KafkaChatHomeService(
@@ -34,7 +35,8 @@ public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceL
     // {
     //   this.offsets[i] = 0l;
     // }
-    this.chatrooms = new Map[numShards];
+    this.kafkaChatRoomServiceMaps = new Map[numShards];
+    this.chatRoomMaps = new Map[numShards];
   }
 
 
@@ -50,6 +52,7 @@ public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceL
       }
 
       int partition = tp.partition();
+      kafkaChatRoomServiceMaps[partition] = new HashMap<>(); // TODO: reuse! Nicht immer alles neu laden
       long unseenOffset = 0; // offsets[partition];
 
       log.info(
@@ -59,7 +62,7 @@ public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceL
           currentOffset);
 
       consumer.seek(tp, unseenOffset);
-      chatrooms[partition]
+      chatRoomMaps[partition]
           .values()
           .stream()
       handlers[partition] = new ChatRoomLoadingMessageHandlingStrategy(tp, currentOffset, unseenOffset);
@@ -101,7 +104,7 @@ public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceL
         (a, b) -> a.addAll(b));
     for (int shard = 0; shard < numShards; shard++)
     {
-      chatrooms[shard] = owned.contains(shard)
+      chatRoomMaps[shard] = owned.contains(shard)
         ? new HashMap<>()
         : null;
     }
@@ -119,25 +122,25 @@ public class KafkaChatHomeService implements ChatHomeService, ConsumerRebalanceL
         }
       })
       .toStream()
-      .forEach(chatroom -> chatrooms[chatroom.getShard()].put(chatroom.getId(), chatroom));
+      .forEach(chatroom -> chatRoomMaps[chatroom.getShard()].put(chatroom.getId(), chatroom));
   }
 
   @Override
   public Mono<ChatRoom> putChatRoom(ChatRoom chatRoom)
   {
-    chatrooms[chatRoom.getShard()].put(chatRoom.getId(), chatRoom);
+    chatRoomMaps[chatRoom.getShard()].put(chatRoom.getId(), chatRoom);
     return Mono.just(chatRoom);
   }
 
   @Override
   public Mono<ChatRoom> getChatRoom(int shard, UUID id)
   {
-    return Mono.justOrEmpty(chatrooms[shard].get(id));
+    return Mono.justOrEmpty(chatRoomMaps[shard].get(id));
   }
 
   @Override
   public Flux<ChatRoom> getChatRooms(int shard)
   {
-    return Flux.fromStream(chatrooms[shard].values().stream());
+    return Flux.fromStream(chatRoomMaps[shard].values().stream());
   }
 }