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