juplo.de Git - demos/kafka/chat/blob - src/test/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaTestUtils.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.ShardingPublisherStrategy;
   5 import lombok.extern.slf4j.Slf4j;
   6 import org.apache.kafka.clients.producer.ProducerRecord;
   7 import org.apache.kafka.common.TopicPartition;
   8 import org.springframework.boot.context.properties.EnableConfigurationProperties;
   9 import org.springframework.boot.test.context.TestConfiguration;
  10 import org.springframework.context.annotation.Bean;
  11 import org.springframework.context.annotation.Import;
  12 import org.springframework.kafka.core.KafkaTemplate;
  13 import org.springframework.kafka.support.SendResult;
  14 import reactor.core.publisher.Mono;
  15
  16 import java.time.Clock;
  17 import java.util.List;
  18
  19
  20 @Slf4j
  21 public class KafkaTestUtils
  22 {
  23   @TestConfiguration
  24   @EnableConfigurationProperties(ChatBackendProperties.class)
  25   @Import(KafkaServicesConfiguration.class)
  26   public static class KafkaTestConfiguration
  27   {
  28     @Bean
  29     public ShardingPublisherStrategy shardingPublisherStrategy()
  30     {
  31       return shard -> Mono.just("MOCKED!");
  32     }
  33
  34     @Bean
  35     public WorkAssignor dataChannelWorkAssignor(
  36         ChatBackendProperties properties,
  37         DataChannel dataChannel)
  38     {
  39       return consumer ->
  40       {
  41         List<TopicPartition> assignedPartitions =
  42             List.of(new TopicPartition(properties.getKafka().getDataChannelTopic(), 2));
  43         consumer.assign(assignedPartitions);
  44         dataChannel.onPartitionsAssigned(assignedPartitions);
  45       };
  46     }
  47
  48     @Bean
  49     public Clock clock()
  50     {
  51       return Clock.systemDefaultZone();
  52     }
  53   }
  54
  55
  56   public static void sendAndLoadStoredData(
  57       KafkaTemplate<String, String> messageTemplate,
  58       String infoTopic,
  59       String dataTopic,
  60       ConsumerTaskRunner consumerTaskRunner)
  61   {
  62     send(messageTemplate, infoTopic, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\": \"5c73531c-6fc4-426c-adcb-afc5c140a0f7\", \"shard\": 2, \"name\": \"FOO\" }", "event_chatroom_created");
  63     send(messageTemplate, dataTopic, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"peter\", \"text\" : \"Hallo, ich heiße Peter!\" }", "event_chatmessage_received");
  64     send(messageTemplate, dataTopic, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"ute\", \"text\" : \"Ich bin Ute...\" }", "event_chatmessage_received");
  65     send(messageTemplate, dataTopic, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 2, \"user\" : \"peter\", \"text\" : \"Willst du mit mir gehen?\" }", "event_chatmessage_received");
  66     send(messageTemplate, dataTopic, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"klaus\", \"text\" : \"Ja? Nein? Vielleicht??\" }", "event_chatmessage_received");
  67
  68     consumerTaskRunner.executeConsumerTasks();
  69   }
  70
  71   private static void send(
  72       KafkaTemplate<String, String> kafkaTemplate,
  73       String topic,
  74       String key,
  75       String value,
  76       String typeId)
  77   {
  78     ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
  79     record.headers().add("__TypeId__", typeId.getBytes());
  80     SendResult<String, String> result = kafkaTemplate.send(record).join();
  81     log.info(
  82         "Sent {}={} to {}",
  83         key,
  84         value,
  85         new TopicPartition(result.getRecordMetadata().topic(), result.getRecordMetadata().partition()));
  86   }
  87
  88   public static void joinConsumerTasks(ConsumerTaskRunner consumerTaskRunner) throws InterruptedException
  89   {
  90     consumerTaskRunner.joinConsumerTasks();
  91   }
  92 }