private final Consumer<String, AbstractMessageTo> consumer;
private final ZoneId zoneId;
private final int numShards;
+ private final Duration pollingInterval;
private final int bufferSize;
private final Clock clock;
private final boolean[] isShardOwned;
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;
Consumer<String, AbstractMessageTo> dataChannelConsumer,
ZoneId zoneId,
int numShards,
+ Duration pollingInterval,
int bufferSize,
Clock clock,
- InfoChannel infoChannel,
+ ChannelMediator channelMediator,
ShardingPublisherStrategy shardingPublisherStrategy)
{
log.debug(
this.producer = producer;
this.zoneId = zoneId;
this.numShards = numShards;
+ this.pollingInterval = pollingInterval;
this.bufferSize = bufferSize;
this.clock = clock;
this.isShardOwned = new boolean[numShards];
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);
});
}
{
try
{
- ConsumerRecords<String, AbstractMessageTo> records = consumer.poll(Duration.ofMinutes(1));
+ ConsumerRecords<String, AbstractMessageTo> records = consumer.poll(pollingInterval);
log.info("Fetched {} messages", records.count());
if (loadInProgress)
return Mono.error(new ShardNotOwnedException(instanceId, shard));
}
- return infoChannel
+ return channelMediator
.getChatRoomInfo(id)
.map(chatRoomInfo ->
chatRoomData[shard].computeIfAbsent(id, this::computeChatRoomData));