juplo.de Git - demos/kafka/chat/blob - src/main/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaServicesApplicationRunner.java

   1 package de.juplo.kafka.chat.backend.implementation.kafka;
   2
   3 import de.juplo.kafka.chat.backend.ChatBackendProperties;
   4 import de.juplo.kafka.chat.backend.implementation.kafka.messages.AbstractMessageTo;
   5 import jakarta.annotation.PreDestroy;
   6 import lombok.RequiredArgsConstructor;
   7 import lombok.extern.slf4j.Slf4j;
   8 import org.apache.kafka.clients.consumer.Consumer;
   9 import org.springframework.beans.factory.annotation.Autowired;
  10 import org.springframework.boot.ApplicationArguments;
  11 import org.springframework.boot.ApplicationRunner;
  12 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
  13 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  14 import org.springframework.stereotype.Component;
  15
  16 import java.util.List;
  17 import java.util.concurrent.CompletableFuture;
  18
  19
  20 @ConditionalOnProperty(
  21     prefix = "chat.backend",
  22     name = "services",
  23     havingValue = "kafka")
  24 @Component
  25 @RequiredArgsConstructor
  26 @Slf4j
  27 public class KafkaServicesApplicationRunner implements ApplicationRunner
  28 {
  29   private final ChatBackendProperties properties;
  30   private final ThreadPoolTaskExecutor taskExecutor;
  31   private final ChatRoomChannel chatRoomChannel;
  32   private final Consumer<String, AbstractMessageTo> chatRoomChannelConsumer;
  33
  34   CompletableFuture<Void> chatRoomChannelConsumerJob;
  35
  36
  37   @Override
  38   public void run(ApplicationArguments args) throws Exception
  39   {
  40     List<String> topics = List.of(properties.getKafka().getChatRoomChannelTopic());
  41     chatRoomChannelConsumer.subscribe(topics, chatRoomChannel);
  42     log.info("Starting the consumer for the ChatRoomChannel");
  43     chatRoomChannelConsumerJob = taskExecutor
  44         .submitCompletable(chatRoomChannel)
  45         .exceptionally(e ->
  46         {
  47           log.error("The consumer for the ChatRoomChannel exited abnormally!", e);
  48           return null;
  49         });
  50   }
  51
  52   @PreDestroy
  53   public void joinChatRoomChannelConsumerJob()
  54   {
  55     log.info("Signaling the consumer of the CahtRoomChannel to quit its work");
  56     chatRoomChannelConsumer.wakeup();
  57     log.info("Waiting for the consumer of the ChatRoomChannel to finish its work");
  58     chatRoomChannelConsumerJob.join();
  59     log.info("Joined the consumer of the ChatRoomChannel");
  60   }
  61 }