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