+ return Mono.justOrEmpty(chatRoomData[shard].get(id));
+ }
+
+ private ChatRoomData computeChatRoomData(UUID chatRoomId, int shard)
+ {
+ ChatRoomData chatRoomData = this.chatRoomData[shard].get(chatRoomId);
+
+ if (chatRoomData != null)
+ {
+ log.info(
+ "Ignoring request to create already existing ChatRoomData for {}",
+ chatRoomId);
+ }
+ else
+ {
+ log.info("Creating ChatRoomData {} with buffer-size {}", chatRoomId, bufferSize);
+ KafkaChatMessageService service = new KafkaChatMessageService(this, chatRoomId);
+ chatRoomData = new ChatRoomData(clock, service, bufferSize);
+ this.chatRoomData[shard].put(chatRoomId, chatRoomData);
+ }
+
+ return chatRoomData;