juplo.de Git - demos/kafka/wordcount/blob - src/test/java/de/juplo/kafka/wordcount/splitter/TestData.java

   1 package de.juplo.kafka.wordcount.splitter;
   2
   3 import de.juplo.kafka.wordcount.counter.TestWord;
   4 import de.juplo.kafka.wordcount.recorder.TestRecording;
   5 import org.apache.kafka.streams.KeyValue;
   6 import org.springframework.util.LinkedMultiValueMap;
   7 import org.springframework.util.MultiValueMap;
   8
   9 import java.time.Duration;
  10 import java.util.function.BiConsumer;
  11 import java.util.stream.Stream;
  12
  13 import static org.assertj.core.api.Assertions.assertThat;
  14 import static org.awaitility.Awaitility.await;
  15
  16
  17 public class TestData
  18 {
  19         static final String PETER = "peter";
  20         static final String KLAUS = "klaus";
  21
  22         static void writeInputData(BiConsumer<String, TestRecording> consumer)
  23         {
  24                 Stream
  25                                 .of(INPUT_MESSAGES)
  26                                 .forEach(kv -> consumer.accept(kv.key, kv.value));
  27         }
  28
  29         static final KeyValue<String, TestRecording>[] INPUT_MESSAGES = new KeyValue[]
  30         {
  31                         new KeyValue<>(
  32                                         PETER,
  33                                         TestRecording.of(PETER, "Hallo Welt!")),
  34                         new KeyValue<>(
  35                                         KLAUS,
  36                                         TestRecording.of(KLAUS, "Müsch gäb's auch!")),
  37                         new KeyValue<>(
  38                                         PETER,
  39                                         TestRecording.of(PETER, "Boäh, echt! ß mal nä Nümmäh!")),
  40         };
  41
  42         static void assertExpectedMessages(MultiValueMap<String, TestWord> receivedMessages)
  43         {
  44                 MultiValueMap<String, TestWord> expected = new LinkedMultiValueMap<>();
  45                 expectedMessages.forEach(keyValue -> expected.add(keyValue.key, keyValue.value));
  46                 await("Received expected messages")
  47                                 .atMost(Duration.ofSeconds(5))
  48                                 .untilAsserted(() -> expected.forEach((user, word) ->
  49                                                 assertThat(receivedMessages.get(user)).containsExactlyElementsOf(word)));
  50         }
  51
  52         static Stream<KeyValue<String, TestWord>> expectedMessages = Stream.of(
  53                         KeyValue.pair(
  54                                         "peter",
  55                                         TestWord.of("peter", "Hallo")),
  56                         KeyValue.pair(
  57                                         "peter",
  58                                         TestWord.of("peter", "Welt")),
  59                         KeyValue.pair(
  60                                         "klaus",
  61                                         TestWord.of("klaus", "Müsch")),
  62                         KeyValue.pair(
  63                                         "klaus",
  64                                         TestWord.of("klaus", "gäb")),
  65                         KeyValue.pair(
  66                                         "klaus",
  67                                         TestWord.of("klaus", "s")),
  68                         KeyValue.pair(
  69                                         "klaus",
  70                                         TestWord.of("klaus", "auch")),
  71                         KeyValue.pair(
  72                                         "peter",
  73                                         TestWord.of("peter", "Boäh")),
  74                         KeyValue.pair(
  75                                         "peter",
  76                                         TestWord.of("peter", "echt")),
  77                         KeyValue.pair(
  78                                         "peter",
  79                                         TestWord.of("peter", "ß")),
  80                         KeyValue.pair(
  81                                         "peter",
  82                                         TestWord.of("peter", "mal")),
  83                         KeyValue.pair(
  84                                         "peter",
  85                                         TestWord.of("peter", "nä")),
  86                         KeyValue.pair(
  87                                         "peter",
  88                                         TestWord.of("peter", "Nümmäh")));
  89 }