From c186a4d189e2bfe1483f11f209b74fed1e7c7ff3 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Fri, 15 Sep 2023 21:35:17 +0200 Subject: [PATCH] WIP --- .../implementation/kafka/DataChannel.java | 18 ++++++++++++------ 1 file changed, 12 insertions(+), 6 deletions(-) diff --git a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/DataChannel.java b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/DataChannel.java index d2d6f300..181f7bf7 100644 --- a/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/DataChannel.java +++ b/src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/DataChannel.java @@ -19,6 +19,7 @@ import reactor.core.publisher.Mono; import java.time.*; import java.util.*; +import java.util.function.Function; import java.util.stream.IntStream; @@ -300,11 +301,16 @@ public class DataChannel implements Runnable, ConsumerRebalanceListener .getChatRoomInfo(id) .map(chatRoomInfo -> chatRoomData[shard].computeIfAbsent( id, - (chatRoomId) -> - { - log.info("Creating ChatRoom {} with buffer-size {}", chatRoomId, bufferSize); - KafkaChatMessageService service = new KafkaChatMessageService(this, chatRoomId); - return new ChatRoomData(clock, service, bufferSize); - })); + computeChatRoomData())); + } + + private Function computeChatRoomData() + { + return (chatRoomId) -> + { + log.info("Creating ChatRoom {} with buffer-size {}", chatRoomId, bufferSize); + KafkaChatMessageService service = new KafkaChatMessageService(this, chatRoomId); + return new ChatRoomData(clock, service, bufferSize); + }; } } -- 2.20.1