import de.juplo.kafka.chat.backend.implementation.kafka.messages.info.EventChatRoomCreated;
import de.juplo.kafka.chat.backend.implementation.kafka.messages.info.EventShardAssigned;
import de.juplo.kafka.chat.backend.implementation.kafka.messages.info.EventShardRevoked;
+import lombok.Getter;
+import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.errors.WakeupException;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.net.URI;
-import java.time.*;
+import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.stream.IntStream;
+@ToString(of = { "topic", "instanceUri" })
@Slf4j
-public class InfoChannel implements Runnable
+public class InfoChannel implements Channel
{
private final String topic;
private final Producer<String, AbstractMessageTo> producer;
private final Consumer<String, AbstractMessageTo> consumer;
+ private final Duration pollingInterval;
private final int numShards;
private final String[] shardOwners;
private final long[] currentOffset;
private final long[] nextOffset;
private final Map<UUID, ChatRoomInfo> chatRoomInfo;
private final String instanceUri;
+ private final ChannelMediator channelMediator;
private boolean running;
+ @Getter
+ private volatile ChannelState channelState = ChannelState.STARTING;
public InfoChannel(
String topic,
Producer<String, AbstractMessageTo> producer,
Consumer<String, AbstractMessageTo> infoChannelConsumer,
- URI instanceUri)
+ Duration pollingInterval,
+ int numShards,
+ URI instanceUri,
+ ChannelMediator channelMediator)
{
log.debug(
"Creating InfoChannel for topic {}",
this.producer = producer;
this.chatRoomInfo = new HashMap<>();
- this.numShards = consumer
- .partitionsFor(topic)
- .size();
+ this.pollingInterval = pollingInterval;
+
+ this.numShards = numShards;
this.shardOwners = new String[numShards];
this.currentOffset = new long[numShards];
this.nextOffset = new long[numShards];
.forEach(partition -> this.nextOffset[partition] = -1l);
this.instanceUri = instanceUri.toASCIIString();
- }
-
- boolean isLoadInProgress()
- {
- return IntStream
- .range(0, numShards)
- .anyMatch(partition -> nextOffset[partition] < currentOffset[partition]);
+ this.channelMediator = channelMediator;
}
+
Mono<ChatRoomInfo> sendChatRoomCreatedEvent(
UUID chatRoomId,
String name,
IntStream
.range(0, numShards)
.forEach(partition -> this.nextOffset[partition] = 0l);
+ channelState = ChannelState.LOAD_IN_PROGRESS;
while (running)
{
try
{
- ConsumerRecords<String, AbstractMessageTo> records = consumer.poll(Duration.ofMinutes(1));
+ ConsumerRecords<String, AbstractMessageTo> records = consumer.poll(pollingInterval);
log.debug("Fetched {} messages", records.count());
- handleMessages(records);
+ for (ConsumerRecord<String, AbstractMessageTo> record : records)
+ {
+ handleMessage(record);
+ updateNextOffset(record.partition(), record.offset() + 1);
+ }
}
catch (WakeupException e)
{
log.info("Received WakeupException, exiting!");
+ channelState = ChannelState.SHUTTING_DOWN;
running = false;
}
}
log.info("Exiting normally");
}
- private void handleMessages(ConsumerRecords<String, AbstractMessageTo> records)
+ private void updateNextOffset(int partition, long nextOffset)
{
- for (ConsumerRecord<String, AbstractMessageTo> record : records)
+ this.nextOffset[partition] = nextOffset;
+ if (channelState == ChannelState.LOAD_IN_PROGRESS)
{
- switch (record.value().getType())
+ boolean loadInProgress = IntStream
+ .range(0, numShards)
+ .anyMatch(shard -> this.nextOffset[shard] < currentOffset[shard]);
+ if (!loadInProgress)
{
- case EVENT_CHATROOM_CREATED:
- EventChatRoomCreated eventChatRoomCreated =
- (EventChatRoomCreated) record.value();
- createChatRoom(eventChatRoomCreated.toChatRoomInfo());
- break;
-
- case EVENT_SHARD_ASSIGNED:
- EventShardAssigned eventShardAssigned =
- (EventShardAssigned) record.value();
- log.info(
- "Shard {} was assigned to {}",
- eventShardAssigned.getShard(),
- eventShardAssigned.getUri());
- shardOwners[eventShardAssigned.getShard()] = eventShardAssigned.getUri();
- break;
-
- case EVENT_SHARD_REVOKED:
- EventShardRevoked eventShardRevoked =
- (EventShardRevoked) record.value();
- log.info(
- "Shard {} was revoked from {}",
- eventShardRevoked.getShard(),
- eventShardRevoked.getUri());
- shardOwners[eventShardRevoked.getShard()] = null;
- break;
-
- default:
- log.debug(
- "Ignoring message for key={} with offset={}: {}",
- record.key(),
- record.offset(),
- record.value());
+ log.info("Loading of info completed! Resuming normal operations...");
+ channelState = ChannelState.READY;
}
+ }
+ }
- nextOffset[record.partition()] = record.offset() + 1;
+ private void handleMessage(ConsumerRecord<String, AbstractMessageTo> record)
+ {
+ switch (record.value().getType())
+ {
+ case EVENT_CHATROOM_CREATED:
+ EventChatRoomCreated eventChatRoomCreated =
+ (EventChatRoomCreated) record.value();
+ createChatRoom(eventChatRoomCreated.toChatRoomInfo());
+ break;
+
+ case EVENT_SHARD_ASSIGNED:
+ EventShardAssigned eventShardAssigned =
+ (EventShardAssigned) record.value();
+ log.info(
+ "Shard {} was assigned to {}",
+ eventShardAssigned.getShard(),
+ eventShardAssigned.getUri());
+ shardOwners[eventShardAssigned.getShard()] = eventShardAssigned.getUri();
+ break;
+
+ case EVENT_SHARD_REVOKED:
+ EventShardRevoked eventShardRevoked =
+ (EventShardRevoked) record.value();
+ log.info(
+ "Shard {} was revoked from {}",
+ eventShardRevoked.getShard(),
+ eventShardRevoked.getUri());
+ shardOwners[eventShardRevoked.getShard()] = null;
+ break;
+
+ default:
+ log.debug(
+ "Ignoring message for key={} with offset={}: {}",
+ record.key(),
+ record.offset(),
+ record.value());
}
}
chatRoomId);
this.chatRoomInfo.put(chatRoomId, chatRoomInfo);
+ this.channelMediator.chatRoomCreated(chatRoomInfo);
}
}
Mono<ChatRoomInfo> getChatRoomInfo(UUID id)
{
+ ChannelState capturedState = channelState;
+ if (capturedState != ChannelState.READY)
+ {
+ return Mono.error(new ChannelNotReadyException(capturedState));
+ }
+
return Mono.fromSupplier(() -> chatRoomInfo.get(id));
}
projects / demos / kafka / chat / blobdiff