* If a new chat-room was created, `InfoChannel` only reacted with the
creation of the according `ChatRoomInfo`-instance.
* The creation of the accompanying `ChatRoomData`-instance through
`DataChannel` was posponed until the new chat-room was accessed the
first time.
* That way, `InfoChannel` did not need to know `DataChannel`, so that a
cyclic dependency could be avoided.
* As a downside, this approach was open to a race-condition: if several
accesses to the newly created chat-room happend in parallel, a
`ConcurrentModificationException` was thrown, since the instance of
`ChatRoomData` was created multiple times in parallel.
* To circumvent the locking, that would be necesarry to evade this race
condition, the approach was refactored, so that `InfoChannel` now
explicitly triggers the creation of the `ChatRoomData`-instance.
* To do so without introducing a cyclic dependency, the class
`ChannelMediator` was introduced, so that `InfoChannel` and `DataChannel`
need not to know each other.
{
@Setter
private InfoChannel infoChannel;
{
@Setter
private InfoChannel infoChannel;
+ @Setter
+ private DataChannel dataChannel;
void shardAssigned(int shard)
void shardAssigned(int shard)
{
return infoChannel.getChatRoomInfo(id);
}
{
return infoChannel.getChatRoomInfo(id);
}
+
+ void chatRoomCreated(ChatRoomInfo chatRoomInfo)
+ {
+ dataChannel.createChatRoomData(chatRoomInfo);
+ }
Message.MessageKey key = Message.MessageKey.of(chatMessageTo.getUser(), chatMessageTo.getId());
Message message = new Message(key, offset, timestamp, chatMessageTo.getText());
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();
KafkaChatMessageService kafkaChatRoomService =
(KafkaChatMessageService) chatRoomData.getChatRoomService();
+ void createChatRoomData(ChatRoomInfo chatRoomInfo)
+ {
+ computeChatRoomData(chatRoomInfo.getId(), chatRoomInfo.getShard());
+ }
+
Mono<ChatRoomData> getChatRoomData(int shard, UUID id)
{
if (loadInProgress)
Mono<ChatRoomData> getChatRoomData(int shard, UUID id)
{
if (loadInProgress)
return Mono.error(new ShardNotOwnedException(instanceId, shard));
}
return Mono.error(new ShardNotOwnedException(instanceId, shard));
}
- return channelMediator
- .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()
}
ConsumerGroupMetadata getConsumerGroupMetadata()
private final long[] nextOffset;
private final Map<UUID, ChatRoomInfo> chatRoomInfo;
private final String instanceUri;
private final long[] nextOffset;
private final Map<UUID, ChatRoomInfo> chatRoomInfo;
private final String instanceUri;
+ private final ChannelMediator channelMediator;
private boolean running;
@Getter
private boolean running;
@Getter
Consumer<String, AbstractMessageTo> infoChannelConsumer,
Duration pollingInterval,
int numShards,
Consumer<String, AbstractMessageTo> infoChannelConsumer,
Duration pollingInterval,
int numShards,
+ URI instanceUri,
+ ChannelMediator channelMediator)
{
log.debug(
"Creating InfoChannel for topic {}",
{
log.debug(
"Creating InfoChannel for topic {}",
.forEach(partition -> this.nextOffset[partition] = -1l);
this.instanceUri = instanceUri.toASCIIString();
.forEach(partition -> this.nextOffset[partition] = -1l);
this.instanceUri = instanceUri.toASCIIString();
+
+ this.channelMediator = channelMediator;
chatRoomId);
this.chatRoomInfo.put(chatRoomId, chatRoomInfo);
chatRoomId);
this.chatRoomInfo.put(chatRoomId, chatRoomInfo);
+ this.channelMediator.chatRoomCreated(chatRoomInfo);
infoChannelConsumer,
properties.getKafka().getPollingInterval(),
properties.getKafka().getNumPartitions(),
infoChannelConsumer,
properties.getKafka().getPollingInterval(),
properties.getKafka().getNumPartitions(),
- properties.getKafka().getInstanceUri());
+ properties.getKafka().getInstanceUri(),
+ channelMediator);
channelMediator.setInfoChannel(infoChannel);
return infoChannel;
}
channelMediator.setInfoChannel(infoChannel);
return infoChannel;
}
ChannelMediator channelMediator,
ShardingPublisherStrategy shardingPublisherStrategy)
{
ChannelMediator channelMediator,
ShardingPublisherStrategy shardingPublisherStrategy)
{
- return new DataChannel(
+ DataChannel dataChannel = new DataChannel(
properties.getInstanceId(),
properties.getKafka().getDataChannelTopic(),
producer,
properties.getInstanceId(),
properties.getKafka().getDataChannelTopic(),
producer,
clock,
channelMediator,
shardingPublisherStrategy);
clock,
channelMediator,
shardingPublisherStrategy);
+ channelMediator.setDataChannel(dataChannel);
+ return dataChannel;