private final long[] currentOffset;
private final long[] nextOffset;
private final Map<UUID, ChatRoomData>[] chatRoomData;
- private final InfoChannel infoChannel;
+ private final ChannelMediator channelMediator;
private final ShardingPublisherStrategy shardingPublisherStrategy;
private boolean running;
Duration pollingInterval,
int bufferSize,
Clock clock,
- InfoChannel infoChannel,
+ ChannelMediator channelMediator,
ShardingPublisherStrategy shardingPublisherStrategy)
{
log.debug(
IntStream
.range(0, numShards)
.forEach(shard -> this.chatRoomData[shard] = new HashMap<>());
- this.infoChannel = infoChannel;
+ this.channelMediator = channelMediator;
this.shardingPublisherStrategy = shardingPublisherStrategy;
}
currentOffset);
consumer.seek(topicPartition, nextOffset[partition]);
- infoChannel.sendShardAssignedEvent(partition);
+ channelMediator.shardAssigned(partition);
shardingPublisherStrategy
.publishOwnership(partition)
.doOnSuccess(instanceId -> log.info(
"Could not publish instance {} as owner of shard {}: {}",
instanceId,
partition,
- throwable))
+ throwable.toString()))
+ .onErrorComplete()
.block();
});
isShardOwned[partition] = false;
nextOffset[partition] = consumer.position(topicPartition);
log.info("Partition revoked: {} - next={}", partition, nextOffset[partition]);
- infoChannel.sendShardRevokedEvent(partition);
+ channelMediator.shardRevoked(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]
- .computeIfAbsent(chatRoomId, this::computeChatRoomData);
+ ChatRoomData chatRoomData = computeChatRoomData(chatRoomId, partition);
KafkaChatMessageService kafkaChatRoomService =
(KafkaChatMessageService) chatRoomData.getChatRoomService();
.toArray();
}
+ void createChatRoomData(ChatRoomInfo chatRoomInfo)
+ {
+ computeChatRoomData(chatRoomInfo.getId(), chatRoomInfo.getShard());
+ }
+
Mono<ChatRoomData> getChatRoomData(int shard, UUID id)
{
if (loadInProgress)
return Mono.error(new ShardNotOwnedException(instanceId, shard));
}
- return infoChannel
- .getChatRoomInfo(id)
- .map(chatRoomInfo ->
- chatRoomData[shard].computeIfAbsent(id, this::computeChatRoomData));
+ return Mono.justOrEmpty(chatRoomData[shard].get(id));
}
- private ChatRoomData computeChatRoomData(UUID chatRoomId)
+ private ChatRoomData computeChatRoomData(UUID chatRoomId, int shard)
{
- log.info("Creating ChatRoom {} with buffer-size {}", chatRoomId, bufferSize);
- KafkaChatMessageService service = new KafkaChatMessageService(this, chatRoomId);
- return new ChatRoomData(clock, service, bufferSize);
+ 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;
}
ConsumerGroupMetadata getConsumerGroupMetadata()