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.ChatBackendProperties;
   4 import de.juplo.kafka.chat.backend.domain.ChatHomeServiceWithShardsTest;
   5 import lombok.extern.slf4j.Slf4j;
   6 import org.apache.kafka.clients.producer.ProducerRecord;
   7 import org.apache.kafka.common.TopicPartition;
   8 import org.junit.jupiter.api.AfterAll;
   9 import org.junit.jupiter.api.BeforeAll;
  10 import org.springframework.beans.factory.annotation.Autowired;
  11 import org.springframework.boot.DefaultApplicationArguments;
  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.context.annotation.Import;
  19 import org.springframework.kafka.core.KafkaTemplate;
  20 import org.springframework.kafka.support.SendResult;
  21 import org.springframework.kafka.test.context.EmbeddedKafka;
  22
  23 import java.time.Clock;
  24 import java.util.List;
  25
  26 import static de.juplo.kafka.chat.backend.domain.ChatHomeServiceWithShardsTest.NUM_SHARDS;
  27 import static de.juplo.kafka.chat.backend.implementation.kafka.KafkaChatHomeServiceTest.TOPIC;
  28
  29
  30 @SpringBootTest(
  31     classes = {
  32         KafkaChatHomeServiceTest.KafkaChatHomeTestConfiguration.class,
  33         KafkaAutoConfiguration.class,
  34         TaskExecutionAutoConfiguration.class,
  35     },
  36     properties = {
  37     "spring.main.allow-bean-definition-overriding=true",
  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=" + NUM_SHARDS,
  44 })
  45 @EmbeddedKafka(topics = { TOPIC }, partitions = 10)
  46 @Slf4j
  47 public class KafkaChatHomeServiceTest extends ChatHomeServiceWithShardsTest
  48 {
  49   final static String TOPIC = "KAFKA_CHAT_HOME_TEST";
  50
  51
  52   @TestConfiguration
  53   @EnableConfigurationProperties(ChatBackendProperties.class)
  54   @Import(KafkaServicesConfiguration.class)
  55   static class KafkaChatHomeTestConfiguration
  56   {
  57     @Bean
  58     KafkaServicesApplicationRunner.WorkAssignor workAssignor(
  59         ChatRoomChannel chatRoomChannel)
  60     {
  61       return consumer ->
  62       {
  63         List<TopicPartition> assignedPartitions =
  64             List.of(new TopicPartition(TOPIC, 2));
  65         consumer.assign(assignedPartitions);
  66         chatRoomChannel.onPartitionsAssigned(assignedPartitions);
  67       };
  68     }
  69
  70     @Bean
  71     Clock clock()
  72     {
  73       return Clock.systemDefaultZone();
  74     }
  75   }
  76
  77
  78   @BeforeAll
  79   public static void sendAndLoadStoredData(@Autowired KafkaTemplate<String, String> messageTemplate)
  80   {
  81     send(messageTemplate, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\": \"5c73531c-6fc4-426c-adcb-afc5c140a0f7\", \"shard\": 2, \"name\": \"FOO\" }", "command_create_chatroom");
  82     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"peter\", \"text\" : \"Hallo, ich heiße Peter!\" }", "event_chatmessage_received");
  83     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"ute\", \"text\" : \"Ich bin Ute...\" }", "event_chatmessage_received");
  84     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 2, \"user\" : \"peter\", \"text\" : \"Willst du mit mir gehen?\" }", "event_chatmessage_received");
  85     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"klaus\", \"text\" : \"Ja? Nein? Vielleicht??\" }", "event_chatmessage_received");
  86   }
  87
  88   static void send(KafkaTemplate<String, String> kafkaTemplate, String key, String value, String typeId)
  89   {
  90     ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, key, value);
  91     record.headers().add("__TypeId__", typeId.getBytes());
  92     SendResult<String, String> result = kafkaTemplate.send(record).join();
  93     log.info(
  94         "Sent {}={} to {}",
  95         key,
  96         value,
  97         new TopicPartition(result.getRecordMetadata().topic(), result.getRecordMetadata().partition()));
  98   }
  99
 100   @AfterAll
 101   static void joinConsumerJob(@Autowired KafkaServicesApplicationRunner applicationRunner)
 102   {
 103     applicationRunner.joinChatRoomChannelConsumerJob();
 104   }
 105 }