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.autoconfigure.kafka.KafkaAutoConfiguration;
  12 import org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration;
  13 import org.springframework.boot.context.properties.EnableConfigurationProperties;
  14 import org.springframework.boot.test.context.SpringBootTest;
  15 import org.springframework.boot.test.context.TestConfiguration;
  16 import org.springframework.context.annotation.Bean;
  17 import org.springframework.context.annotation.Import;
  18 import org.springframework.kafka.core.KafkaTemplate;
  19 import org.springframework.kafka.support.SendResult;
  20 import org.springframework.kafka.test.context.EmbeddedKafka;
  21
  22 import java.time.Clock;
  23 import java.util.List;
  24
  25 import static de.juplo.kafka.chat.backend.domain.ChatHomeServiceWithShardsTest.NUM_SHARDS;
  26 import static de.juplo.kafka.chat.backend.implementation.kafka.KafkaChatHomeServiceTest.TOPIC;
  27
  28
  29 @SpringBootTest(
  30     classes = {
  31         KafkaChatHomeServiceTest.KafkaChatHomeTestConfiguration.class,
  32         KafkaAutoConfiguration.class,
  33         TaskExecutionAutoConfiguration.class,
  34     },
  35     properties = {
  36     "spring.main.allow-bean-definition-overriding=true",
  37     "chat.backend.services=kafka",
  38     "chat.backend.kafka.client-id-PREFIX=TEST",
  39     "chat.backend.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
  40     "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
  41     "chat.backend.kafka.chatroom-channel-topic=" + TOPIC,
  42     "chat.backend.kafka.num-partitions=" + NUM_SHARDS,
  43 })
  44 @EmbeddedKafka(topics = { TOPIC }, partitions = 10)
  45 @Slf4j
  46 public class KafkaChatHomeServiceTest extends ChatHomeServiceWithShardsTest
  47 {
  48   final static String TOPIC = "KAFKA_CHAT_HOME_TEST";
  49
  50
  51   @TestConfiguration
  52   @EnableConfigurationProperties(ChatBackendProperties.class)
  53   @Import(KafkaServicesConfiguration.class)
  54   static class KafkaChatHomeTestConfiguration
  55   {
  56     @Bean
  57     ConsumerTaskExecutor.WorkAssignor workAssignor(
  58         ChatRoomChannel chatRoomChannel)
  59     {
  60       return consumer ->
  61       {
  62         List<TopicPartition> assignedPartitions =
  63             List.of(new TopicPartition(TOPIC, 2));
  64         consumer.assign(assignedPartitions);
  65         chatRoomChannel.onPartitionsAssigned(assignedPartitions);
  66       };
  67     }
  68
  69     @Bean
  70     Clock clock()
  71     {
  72       return Clock.systemDefaultZone();
  73     }
  74   }
  75
  76
  77   @BeforeAll
  78   public static void sendAndLoadStoredData(
  79       @Autowired ConsumerTaskExecutor consumerTaskExecutor,
  80       @Autowired KafkaTemplate<String, String> messageTemplate)
  81   {
  82     send(messageTemplate, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\": \"5c73531c-6fc4-426c-adcb-afc5c140a0f7\", \"shard\": 2, \"name\": \"FOO\" }", "command_create_chatroom");
  83     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"peter\", \"text\" : \"Hallo, ich heiße Peter!\" }", "event_chatmessage_received");
  84     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"ute\", \"text\" : \"Ich bin Ute...\" }", "event_chatmessage_received");
  85     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 2, \"user\" : \"peter\", \"text\" : \"Willst du mit mir gehen?\" }", "event_chatmessage_received");
  86     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"klaus\", \"text\" : \"Ja? Nein? Vielleicht??\" }", "event_chatmessage_received");
  87
  88     consumerTaskExecutor.executeConsumerTask();
  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 ConsumerTaskExecutor consumerTaskExecutor)
 105   {
 106     consumerTaskExecutor.joinConsumerTaskJob();
 107   }
 108 }