juplo.de Git - demos/kafka/chat/blob - src/test/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaChatHomeServiceTest.java

   1 package de.juplo.kafka.chat.backend.implementation.kafka;
   2
   3 import de.juplo.kafka.chat.backend.domain.ChatHomeServiceWithShardsTest;
   4 import lombok.extern.slf4j.Slf4j;
   5 import org.apache.kafka.clients.consumer.Consumer;
   6 import org.junit.jupiter.api.AfterAll;
   7 import org.junit.jupiter.api.BeforeAll;
   8 import org.springframework.beans.factory.annotation.Autowired;
   9 import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
  10 import org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration;
  11 import org.springframework.kafka.core.KafkaTemplate;
  12 import org.springframework.kafka.test.context.EmbeddedKafka;
  13 import org.springframework.test.context.ContextConfiguration;
  14 import org.springframework.test.context.TestPropertySource;
  15
  16 import static de.juplo.kafka.chat.backend.domain.ChatHomeServiceWithShardsTest.NUM_SHARDS;
  17 import static de.juplo.kafka.chat.backend.implementation.kafka.KafkaChatHomeServiceTest.DATA_TOPIC;
  18 import static de.juplo.kafka.chat.backend.implementation.kafka.KafkaChatHomeServiceTest.INFO_TOPIC;
  19
  20
  21 @ContextConfiguration(classes = {
  22         KafkaTestUtils.KafkaTestConfiguration.class,
  23         KafkaAutoConfiguration.class,
  24         TaskExecutionAutoConfiguration.class,
  25     })
  26 @TestPropertySource(properties = {
  27         "chat.backend.services=kafka",
  28         "chat.backend.kafka.client-id-PREFIX=TEST",
  29         "chat.backend.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
  30         "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
  31         "chat.backend.kafka.info-channel-topic=" + INFO_TOPIC,
  32         "chat.backend.kafka.data-channel-topic=" + DATA_TOPIC,
  33         "chat.backend.kafka.num-partitions=" + NUM_SHARDS,
  34 })
  35 @EmbeddedKafka(
  36     topics = { INFO_TOPIC, DATA_TOPIC },
  37     partitions = NUM_SHARDS)
  38 @Slf4j
  39 public class KafkaChatHomeServiceTest extends ChatHomeServiceWithShardsTest
  40 {
  41   final static String INFO_TOPIC = "KAFKA_CHAT_HOME_TEST_INFO";
  42   final static String DATA_TOPIC = "KAFKA_CHAT_HOME_TEST_DATA";
  43
  44
  45   @BeforeAll
  46   static void sendAndLoadStoredData(
  47       @Autowired KafkaTemplate<String, String> messageTemplate,
  48       @Autowired ChannelTaskRunner channelTaskRunner)
  49   {
  50     KafkaTestUtils.sendAndLoadStoredData(
  51         messageTemplate,
  52         INFO_TOPIC,
  53         DATA_TOPIC,
  54         channelTaskRunner);
  55   }
  56
  57   @AfterAll
  58   static void joinConsumerTasks(
  59       @Autowired Consumer dataChannelConsumer,
  60       @Autowired Consumer infoChannelConsumer,
  61       @Autowired ChannelTaskRunner channelTaskRunner)
  62       throws InterruptedException
  63   {
  64     dataChannelConsumer.wakeup();
  65     infoChannelConsumer.wakeup();
  66     channelTaskRunner.joinChannels();
  67   }
  68 }