From cce6ad12b547bcd2c371f2d97db7c617661e921a Mon Sep 17 00:00:00 2001
From: Kai Moritz <kai@juplo.de>
Date: Fri, 18 Aug 2023 16:10:45 +0200
Subject: [PATCH] NG
---
.../persistence/kafka/ChatMessageChannel.java | 8 ++++++++
.../backend/persistence/kafka/KafkaChatHome.java | 15 ++++++++++++---
.../persistence/kafka/KafkaChatRoomFactory.java | 4 ++--
3 files changed, 22 insertions(+), 5 deletions(-)
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 b473a418..7f885e6f 100644
--- 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
@@ -362,4 +362,12 @@ public class ChatMessageChannel implements Runnable, ConsumerRebalanceListener
{
return Mono.justOrEmpty(chatrooms[shard].get(id));
}
+
+ Flux<ChatRoom> getChatRooms()
+ {
+ return Flux
+ .fromStream(IntStream.range(0, numShards).mapToObj(i -> Integer.valueOf(i)))
+ .filter(shard -> isShardOwned[shard])
+ .flatMap(shard -> Flux.fromIterable(chatrooms[shard].values()));
+ }
}
diff --git a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatHome.java b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatHome.java
index 2bccc3ad..71893624 100644
--- a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatHome.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatHome.java
@@ -4,6 +4,7 @@ import de.juplo.kafka.chat.backend.domain.ChatHome;
import de.juplo.kafka.chat.backend.domain.ChatRoom;
import de.juplo.kafka.chat.backend.domain.ChatRoomInfo;
import de.juplo.kafka.chat.backend.domain.ShardingStrategy;
+import de.juplo.kafka.chat.backend.persistence.KafkaLikeShardingStrategy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;
@@ -16,11 +17,19 @@ import java.util.*;
@Slf4j
public class KafkaChatHome implements ChatHome
{
- private final ShardingStrategy shardingStrategy;
- private final ChatRoomChannel chatRoomChannel;
+ private final KafkaLikeShardingStrategy shardingStrategy;
private final ChatMessageChannel chatMessageChanel;
+ public KafkaChatHome(
+ int numPartitions,
+ ChatMessageChannel chatMessageChannel)
+ {
+ this.shardingStrategy = new KafkaLikeShardingStrategy(numPartitions);
+ this.chatMessageChanel = chatMessageChannel;
+ }
+
+
@Override
public Mono<ChatRoom> getChatRoom(UUID id)
{
@@ -38,6 +47,6 @@ public class KafkaChatHome implements ChatHome
@Override
public Flux<ChatRoom> getChatRooms()
{
- return chatRoomChannel.getChatRooms();
+ return chatMessageChanel.getChatRooms();
}
}
diff --git a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatRoomFactory.java b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatRoomFactory.java
index 825f16eb..c46529d8 100644
--- a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatRoomFactory.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatRoomFactory.java
@@ -13,12 +13,12 @@ import java.util.UUID;
@Slf4j
public class KafkaChatRoomFactory implements ChatRoomFactory
{
- private final ChatRoomChannel chatRoomChannel;
+ private final ChatMessageChannel chatMessageChannel;
@Override
public Mono<ChatRoomInfo> createChatRoom(UUID id, String name)
{
log.info("Sending create-request for chat rooom: id={}, name={}");
- return chatRoomChannel.sendCreateChatRoomRequest(id, name);
+ return chatMessageChannel.sendCreateChatRoomRequest(id, name);
}
}
--
2.20.1