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

   1 package de.juplo.kafka.chat.backend.persistence.kafka;
   2
   3 import de.juplo.kafka.chat.backend.ChatBackendProperties;
   4 import de.juplo.kafka.chat.backend.persistence.kafka.messages.AbstractMessageTo;
   5 import jakarta.annotation.PreDestroy;
   6 import lombok.extern.slf4j.Slf4j;
   7 import org.apache.kafka.clients.consumer.Consumer;
   8 import org.springframework.beans.factory.annotation.Autowired;
   9 import org.springframework.boot.ApplicationArguments;
  10 import org.springframework.boot.ApplicationRunner;
  11 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
  12 import org.springframework.context.ConfigurableApplicationContext;
  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 @Slf4j
  26 public class KafkaServicesApplicationRunner implements ApplicationRunner
  27 {
  28   @Autowired
  29   ChatBackendProperties properties;
  30
  31   @Autowired
  32   ThreadPoolTaskExecutor taskExecutor;
  33   @Autowired
  34   ConfigurableApplicationContext context;
  35
  36   @Autowired
  37   ChatRoomChannel chatRoomChannel;
  38   @Autowired
  39   Consumer<String, AbstractMessageTo> chatRoomChannelConsumer;
  40
  41   CompletableFuture<Void> chatRoomChannelConsumerJob;
  42
  43
  44   @Override
  45   public void run(ApplicationArguments args) throws Exception
  46   {
  47     List<String> topics = List.of(properties.getKafka().getChatRoomChannelTopic());
  48     chatRoomChannelConsumer.subscribe(topics, chatRoomChannel);
  49     log.info("Starting the consumer for the ChatRoomChannel");
  50     chatRoomChannelConsumerJob = taskExecutor
  51         .submitCompletable(chatRoomChannel)
  52         .exceptionally(e ->
  53         {
  54           log.error("The consumer for the ChatRoomChannel exited abnormally!", e);
  55           return null;
  56         });
  57   }
  58
  59   @PreDestroy
  60   public void joinChatRoomChannelConsumerJob()
  61   {
  62     log.info("Signaling the consumer of the CahtRoomChannel to quit its work");
  63     chatRoomChannelConsumer.wakeup();
  64     log.info("Waiting for the consumer of the ChatRoomChannel to finish its work");
  65     chatRoomChannelConsumerJob.join();
  66     log.info("Joined the consumer of the ChatRoomChannel");
  67   }
  68 }