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 jakarta.annotation.PreDestroy;
   4 import lombok.extern.slf4j.Slf4j;
   5 import org.springframework.beans.factory.annotation.Autowired;
   6 import org.springframework.boot.ApplicationArguments;
   7 import org.springframework.boot.ApplicationRunner;
   8 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
   9 import org.springframework.context.ConfigurableApplicationContext;
  10 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  11 import org.springframework.stereotype.Component;
  12
  13 import java.util.concurrent.CompletableFuture;
  14
  15
  16 @ConditionalOnProperty(
  17     prefix = "chat.backend",
  18     name = "services",
  19     havingValue = "kafka")
  20 @Component
  21 @Slf4j
  22 public class KafkaServicesApplicationRunner implements ApplicationRunner
  23 {
  24   @Autowired
  25   ThreadPoolTaskExecutor taskExecutor;
  26   @Autowired
  27   ConfigurableApplicationContext context;
  28
  29   @Autowired
  30   ChatMessageChannel chatMessageChannel;
  31   @Autowired
  32   ChatRoomChannel chatRoomChannel;
  33
  34   CompletableFuture<Void> chatRoomChannelConsumerJob;
  35   CompletableFuture<Void> chatMessageChannelConsumerJob;
  36
  37
  38   @Override
  39   public void run(ApplicationArguments args) throws Exception
  40   {
  41     log.info("Starting the consumer for the ChatRoomChannel");
  42     chatRoomChannelConsumerJob = taskExecutor
  43         .submitCompletable(chatRoomChannel)
  44         .exceptionally(e ->
  45         {
  46           log.error("The consumer for the ChatRoomChannel exited abnormally!", e);
  47           return null;
  48         });
  49     log.info("Starting the consumer for the ChatMessageChannel");
  50     chatMessageChannelConsumerJob = taskExecutor
  51         .submitCompletable(chatMessageChannel)
  52         .exceptionally(e ->
  53         {
  54           log.error("The consumer for the ChatMessageChannel exited abnormally!", e);
  55           return null;
  56         });
  57   }
  58
  59   @PreDestroy
  60   public void joinChatRoomChannelConsumerJob()
  61   {
  62     log.info("Waiting for the consumer of the ChatRoomChannel to finish its work");
  63     chatRoomChannelConsumerJob.join();
  64     log.info("Joined the consumer of the ChatRoomChannel");
  65   }
  66
  67   @PreDestroy
  68   public void joinChatMessageChannelConsumerJob()
  69   {
  70     log.info("Waiting for the consumer of the ChatMessageChannel to finish its work");
  71     chatMessageChannelConsumerJob.join();
  72     log.info("Joined the consumer of the ChatMessageChannel");
  73   }
  74 }