- UUID id = chatRoomInfo.getId();
- log.info("Creating ChatRoom {} with buffer-size {}", id, bufferSize);
- KafkaChatMessageService service = new KafkaChatMessageService(this, id);
- ChatRoomData chatRoomData = new ChatRoomData(clock, service, bufferSize);
- putChatRoom(
- chatRoomInfo.getId(),
- chatRoomInfo.getName(),
- chatRoomInfo.getShard(),
- chatRoomData);
- }
-
- private void loadChatMessage(
- UUID chatRoomId,
- LocalDateTime timestamp,
- long offset,
- EventChatMessageReceivedTo chatMessageTo,
- int partition)
- {
- Message.MessageKey key = Message.MessageKey.of(chatMessageTo.getUser(), chatMessageTo.getId());
- Message message = new Message(key, offset, timestamp, chatMessageTo.getText());
-
- ChatRoomData chatRoomData = this.chatRoomData[partition].get(chatRoomId);
- KafkaChatMessageService kafkaChatRoomService =
- (KafkaChatMessageService) chatRoomData.getChatRoomService();
-
- kafkaChatRoomService.persistMessage(message);
- }
-
- private boolean isLoadingCompleted()
- {
- return IntStream
- .range(0, numShards)
- .filter(shard -> isShardOwned[shard])
- .allMatch(shard -> nextOffset[shard] >= currentOffset[shard]);
- }