juplo.de Git - demos/kafka/chat/blob - src/test/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatHomeTest.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.domain.ChatHomeTest;
   5 import lombok.extern.slf4j.Slf4j;
   6 import org.apache.kafka.clients.consumer.Consumer;
   7 import org.apache.kafka.clients.producer.ProducerRecord;
   8 import org.apache.kafka.common.TopicPartition;
   9 import org.junit.jupiter.api.AfterAll;
  10 import org.junit.jupiter.api.BeforeAll;
  11 import org.springframework.beans.factory.annotation.Autowired;
  12 import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
  13 import org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration;
  14 import org.springframework.boot.context.properties.EnableConfigurationProperties;
  15 import org.springframework.boot.test.context.SpringBootTest;
  16 import org.springframework.boot.test.context.TestConfiguration;
  17 import org.springframework.context.annotation.Bean;
  18 import org.springframework.kafka.core.KafkaTemplate;
  19 import org.springframework.kafka.support.SendResult;
  20 import org.springframework.kafka.test.context.EmbeddedKafka;
  21 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  22
  23 import java.time.Clock;
  24 import java.util.List;
  25 import java.util.concurrent.CompletableFuture;
  26
  27 import static de.juplo.kafka.chat.backend.persistence.kafka.KafkaChatHomeTest.TOPIC;
  28
  29
  30 @SpringBootTest(
  31     classes = {
  32         KafkaChatHomeTest.KafkaChatHomeTestConfiguration.class,
  33         KafkaServicesConfiguration.class,
  34         KafkaAutoConfiguration.class,
  35         TaskExecutionAutoConfiguration.class,
  36     },
  37     properties = {
  38     "chat.backend.services=kafka",
  39     "chat.backend.kafka.client-id-PREFIX=TEST",
  40     "chat.backend.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
  41     "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
  42     "chat.backend.kafka.chatroom-channel-topic=" + TOPIC,
  43     "chat.backend.kafka.num-partitions=10",
  44 })
  45 @EmbeddedKafka(topics = { TOPIC }, partitions = 10)
  46 @Slf4j
  47 public class KafkaChatHomeTest extends ChatHomeTest
  48 {
  49   final static String TOPIC = "KAFKA_CHAT_HOME_TEST";
  50
  51   static CompletableFuture<Void> CONSUMER_JOB;
  52
  53
  54   @TestConfiguration
  55   @EnableConfigurationProperties(ChatBackendProperties.class)
  56   static class KafkaChatHomeTestConfiguration
  57   {
  58     @Bean
  59     Clock clock()
  60     {
  61       return Clock.systemDefaultZone();
  62     }
  63   }
  64
  65
  66   @BeforeAll
  67   public static void sendAndLoadStoredData(
  68       @Autowired KafkaTemplate<String, String> messageTemplate,
  69       @Autowired Consumer chatRoomChannelConsumer,
  70       @Autowired ThreadPoolTaskExecutor taskExecutor,
  71       @Autowired ChatRoomChannel chatRoomChannel)
  72   {
  73     send(messageTemplate, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\": \"5c73531c-6fc4-426c-adcb-afc5c140a0f7\", \"shard\": 2, \"name\": \"FOO\" }", "command_create_chatroom");
  74     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"peter\", \"text\" : \"Hallo, ich heiße Peter!\" }", "event_chatmessage_received");
  75     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"ute\", \"text\" : \"Ich bin Ute...\" }", "event_chatmessage_received");
  76     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 2, \"user\" : \"peter\", \"text\" : \"Willst du mit mir gehen?\" }", "event_chatmessage_received");
  77     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"klaus\", \"text\" : \"Ja? Nein? Vielleicht??\" }", "event_chatmessage_received");
  78
  79     List<TopicPartition> assignedPartitions = List.of(new TopicPartition(TOPIC, 2));
  80     chatRoomChannelConsumer.assign(assignedPartitions);
  81     chatRoomChannel.onPartitionsAssigned(assignedPartitions);
  82     CONSUMER_JOB = taskExecutor
  83         .submitCompletable(chatRoomChannel)
  84         .exceptionally(e ->
  85         {
  86           log.error("The consumer for the ChatRoomChannel exited abnormally!", e);
  87           return null;
  88         });
  89   }
  90
  91   static void send(KafkaTemplate<String, String> kafkaTemplate, String key, String value, String typeId)
  92   {
  93     ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, key, value);
  94     record.headers().add("__TypeId__", typeId.getBytes());
  95     SendResult<String, String> result = kafkaTemplate.send(record).join();
  96     log.info(
  97         "Sent {}={} to {}",
  98         key,
  99         value,
 100         new TopicPartition(result.getRecordMetadata().topic(), result.getRecordMetadata().partition()));
 101   }
 102
 103   @AfterAll
 104   static void joinConsumerJob(@Autowired Consumer chatRoomChannelConsumer)
 105   {
 106     log.info("Signaling the consumer of the CahtRoomChannel to quit its work");
 107     chatRoomChannelConsumer.wakeup();
 108     log.info("Waiting for the consumer of the ChatRoomChannel to finish its work");
 109     CONSUMER_JOB.join();
 110     log.info("Joined the consumer of the ChatRoomChannel");
 111   }
 112 }