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

   1 package de.juplo.kafka.wordcount.counter;
   2
   3 import org.apache.kafka.common.header.Header;
   4 import org.apache.kafka.common.header.Headers;
   5 import org.apache.kafka.streams.KeyValue;
   6 import org.springframework.util.LinkedMultiValueMap;
   7 import org.springframework.util.MultiValueMap;
   8
   9 import java.util.Map;
  10 import java.util.Properties;
  11 import java.util.function.BiConsumer;
  12 import java.util.stream.Collectors;
  13 import java.util.stream.Stream;
  14
  15 import static org.assertj.core.api.Assertions.assertThat;
  16
  17
  18 class TestData
  19 {
  20         static void writeInputData(BiConsumer<String, Word> consumer)
  21         {
  22                 consumer.accept(
  23                                 "peter",
  24                                 Word.of("peter","Hallo"));
  25                 consumer.accept(
  26                                 "klaus",
  27                                 Word.of("klaus","Müsch"));
  28                 consumer.accept(
  29                                 "peter",
  30                                 Word.of("peter","Welt"));
  31                 consumer.accept(
  32                                 "klaus",
  33                                 Word.of("klaus","Müsch"));
  34                 consumer.accept(
  35                                 "klaus",
  36                                 Word.of("klaus","s"));
  37                 consumer.accept(
  38                                 "peter",
  39                                 Word.of("peter","Boäh"));
  40                 consumer.accept(
  41                                 "peter",
  42                                 Word.of("peter","Welt"));
  43                 consumer.accept(
  44                                 "peter",
  45                                 Word.of("peter","Boäh"));
  46                 consumer.accept(
  47                                 "klaus",
  48                                 Word.of("klaus","s"));
  49                 consumer.accept(
  50                                 "peter",
  51                                 Word.of("peter","Boäh"));
  52                 consumer.accept(
  53                                 "klaus",
  54                                 Word.of("klaus","s"));
  55         }
  56
  57         static void assertExpectedResult(MultiValueMap<Word, WordCounter> receivedMessages)
  58         {
  59                 expectedMessages.forEach(
  60                                 (word, counter) ->
  61                                                 assertThat(receivedMessages.get(word))
  62                                                                 .containsExactlyElementsOf(counter));
  63         }
  64
  65         static KeyValue<Word, WordCounter>[] expectedMessagesArray = new KeyValue[]
  66         {
  67                         KeyValue.pair(
  68                                         Word.of("peter","Hallo"),
  69                                         WordCounter.of("peter","Hallo",1)),
  70                         KeyValue.pair(
  71                                         Word.of("klaus","Müsch"),
  72                                         WordCounter.of("klaus","Müsch",1)),
  73                         KeyValue.pair(
  74                                         Word.of("peter","Welt"),
  75                                         WordCounter.of("peter","Welt",1)),
  76                         KeyValue.pair(
  77                                         Word.of("klaus","Müsch"),
  78                                         WordCounter.of("klaus","Müsch",2)),
  79                         KeyValue.pair(
  80                                         Word.of("klaus","s"),
  81                                         WordCounter.of("klaus","s",1)),
  82                         KeyValue.pair(
  83                                         Word.of("peter","Boäh"),
  84                                         WordCounter.of("peter","Boäh",1)),
  85                         KeyValue.pair(
  86                                         Word.of("peter","Welt"),
  87                                         WordCounter.of("peter","Welt",2)),
  88                         KeyValue.pair(
  89                                         Word.of("peter","Boäh"),
  90                                         WordCounter.of("peter","Boäh",2)),
  91                         KeyValue.pair(
  92                                         Word.of("klaus","s"),
  93                                         WordCounter.of("klaus","s",2)),
  94                         KeyValue.pair(
  95                                         Word.of("peter","Boäh"),
  96                                         WordCounter.of("peter","Boäh",3)),
  97                         KeyValue.pair(
  98                                         Word.of("klaus","s"),
  99                                         WordCounter.of("klaus","s",3)),
 100         };
 101
 102         static MultiValueMap<Word, WordCounter> expectedMessages;
 103         static
 104         {
 105                 expectedMessages = new LinkedMultiValueMap<>();
 106                 Stream
 107                                 .of(expectedMessagesArray)
 108                                 .forEach(keyValue -> expectedMessages.add(keyValue.key, keyValue.value));
 109         }
 110
 111         static Map<String, Object> convertToMap(Properties properties)
 112         {
 113                 return properties
 114                                 .entrySet()
 115                                 .stream()
 116                                 .collect(
 117                                                 Collectors.toMap(
 118                                                                 entry -> (String)entry.getKey(),
 119                                                                 entry -> entry.getValue()
 120                                                 ));
 121         }
 122
 123         static String parseHeader(Headers headers, String key)
 124         {
 125                 Header header = headers.lastHeader(key);
 126                 if (header == null)
 127                 {
 128                         return key + "=null";
 129                 }
 130                 else
 131                 {
 132                         return key + "=" + new String(header.value());
 133                 }
 134         }
 135 }