import de.juplo.kafka.chat.backend.persistence.KafkaLikeShardingStrategy;
import de.juplo.kafka.chat.backend.persistence.inmemory.InMemoryChatHomeService;
import de.juplo.kafka.chat.backend.persistence.inmemory.InMemoryChatRoomFactory;
+import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.producer.Producer;
@Autowired
ChatMessageChannel chatMessageChannel;
+ @Autowired
+ ChatRoomChannel chatRoomChannel;
- CompletableFuture<Optional<Exception>> chatRoomChannelConsumerJob;
- CompletableFuture<Optional<Exception>> chatMessageChannelConsumerJob;
+ CompletableFuture<Void> chatRoomChannelConsumerJob;
+ CompletableFuture<Void> chatMessageChannelConsumerJob;
@Override
public void run(ApplicationArguments args) throws Exception
{
log.info("Starting the consumer for the ChatRoomChannel");
- chatRoomChannelConsumerJob = taskExecutor.submitCompletable(chatMessageChannel);
- chatRoomChannelConsumerJob.thenAccept(exceptionOptional ->
- {
- exceptionOptional.ifPresent();
- log.info("SimpleConsumer exited normally, exit-status: {}", exitStatus);
- SpringApplication.exit(context, () -> exitStatus);
- },
- t ->
+ chatRoomChannelConsumerJob = taskExecutor
+ .submitCompletable(chatRoomChannel)
+ .exceptionally(e ->
+ {
+ log.error("The consumer for the ChatRoomChannel exited abnormally!", e);
+ return null;
+ });
+ log.info("Starting the consumer for the ChatMessageChannel");
+ chatMessageChannelConsumerJob = taskExecutor
+ .submitCompletable(chatMessageChannel)
+ .exceptionally(e ->
{
- log.error("SimpleConsumer exited abnormally!", t);
- SpringApplication.exit(context, () -> 2);
+ log.error("The consumer for the ChatMessageChannel exited abnormally!", e);
+ return null;
});
}
@PreDestroy
- public void shutdown() throws ExecutionException, InterruptedException
+ public void joinChatRoomChannelConsumerJob()
+ {
+ log.info("Waiting for the consumer of the ChatRoomChannel to finish its work");
+ chatRoomChannelConsumerJob.join();
+ log.info("Joined the consumer of the ChatRoomChannel");
+ }
+
+ @PreDestroy
+ public void joinChatMessageChannelConsumerJob()
{
- log.info("Signaling SimpleConsumer to quit its work");
- kafkaConsumer.wakeup();
- log.info("Waiting for SimpleConsumer to finish its work");
- consumerJob.get();
- log.info("SimpleConsumer finished its work");
+ log.info("Waiting for the consumer of the ChatMessageChannel to finish its work");
+ chatMessageChannelConsumerJob.join();
+ log.info("Joined the consumer of the ChatMessageChannel");
}