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