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

   1 package de.juplo.kafka.wordcount.top10;
   2
   3 import de.juplo.kafka.wordcount.counter.TestCounter;
   4 import de.juplo.kafka.wordcount.counter.TestWord;
   5 import de.juplo.kafka.wordcount.query.TestEntry;
   6 import de.juplo.kafka.wordcount.query.TestRanking;
   7 import de.juplo.kafka.wordcount.query.TestUser;
   8 import org.apache.kafka.streams.KeyValue;
   9 import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
  10 import org.springframework.util.LinkedMultiValueMap;
  11 import org.springframework.util.MultiValueMap;
  12
  13 import java.util.Arrays;
  14 import java.util.stream.Stream;
  15
  16 import static org.assertj.core.api.Assertions.assertThat;
  17
  18
  19 class TestData
  20 {
  21         static final String TYPE_COUNTER = "COUNTER";
  22
  23         static final TestUser PETER = TestUser.of("peter");
  24         static final TestUser KLAUS = TestUser.of("klaus");
  25
  26         static final Stream<KeyValue<TestWord, TestCounter>> getInputMessages()
  27         {
  28                 return Stream.of(INPUT_MESSAGES);
  29         }
  30
  31         private static final KeyValue<TestWord, TestCounter>[] INPUT_MESSAGES = new KeyValue[]
  32         {
  33                         new KeyValue<>(
  34                                         TestWord.of(TYPE_COUNTER, PETER.getUser(),"Hallo"),
  35                                         TestCounter.of("Hallo",1)),
  36                         new KeyValue<>(
  37                                         TestWord.of(TYPE_COUNTER, KLAUS.getUser(),"Müsch"),
  38                                         TestCounter.of("Müsch",1)),
  39                         new KeyValue<>(
  40                                         TestWord.of(TYPE_COUNTER, PETER.getUser(),"Welt"),
  41                                         TestCounter.of("Welt",1)),
  42                         new KeyValue<>(
  43                                         TestWord.of(TYPE_COUNTER, KLAUS.getUser(),"Müsch"),
  44                                         TestCounter.of("Müsch",2)),
  45                         new KeyValue<>(
  46                                         TestWord.of(TYPE_COUNTER, KLAUS.getUser(),"s"),
  47                                         TestCounter.of("s",1)),
  48                         new KeyValue<>(
  49                                         TestWord.of(TYPE_COUNTER, PETER.getUser(),"Boäh"),
  50                                         TestCounter.of("Boäh",1)),
  51                         new KeyValue<>(
  52                                         TestWord.of(TYPE_COUNTER, PETER.getUser(),"Welt"),
  53                                         TestCounter.of("Welt",2)),
  54                         new KeyValue<>(
  55                                         TestWord.of(TYPE_COUNTER, PETER.getUser(),"Boäh"),
  56                                         TestCounter.of("Boäh",2)),
  57                         new KeyValue<>(
  58                                         TestWord.of(TYPE_COUNTER, KLAUS.getUser(),"s"),
  59                                         TestCounter.of("s",2)),
  60                         new KeyValue<>(
  61                                         TestWord.of(TYPE_COUNTER, PETER.getUser(),"Boäh"),
  62                                         TestCounter.of("Boäh",3)),
  63                         new KeyValue<>(
  64                                         TestWord.of(TYPE_COUNTER, KLAUS.getUser(),"s"),
  65                                         TestCounter.of("s",3)),
  66         };
  67
  68         static void assertExpectedMessages(MultiValueMap<TestUser, TestRanking> receivedMessages)
  69         {
  70                 expectedMessages().forEach(
  71                                 (user, rankings) ->
  72                                                 assertThat(receivedMessages.get(user))
  73                                                                 .containsExactlyElementsOf(rankings));
  74         }
  75
  76         static void assertExpectedState(ReadOnlyKeyValueStore<User, Ranking> store)
  77         {
  78                 assertRankingEqualsRankingFromLastMessage(PETER, store.get(userOf(PETER)));
  79                 assertRankingEqualsRankingFromLastMessage(KLAUS, store.get(userOf(KLAUS)));
  80         }
  81
  82         private static User userOf(TestUser user)
  83         {
  84                 return User.of(user.getUser());
  85         }
  86
  87         static void assertExpectedNumberOfMessagesForUsers(MultiValueMap<TestUser, TestRanking> receivedMessages)
  88         {
  89                 assertThat(countMessagesForUser(PETER, receivedMessages));
  90                 assertThat(countMessagesForUser(KLAUS, receivedMessages));
  91         }
  92
  93         private static int countMessagesForUser(TestUser user, MultiValueMap<TestUser, TestRanking> messagesForUsers)
  94         {
  95                 return messagesForUsers.get(user) == null
  96                                 ? 0
  97                                 : messagesForUsers.get(user).size();
  98         }
  99
 100
 101         static void assertExpectedLastMessagesForUsers(MultiValueMap<TestUser, TestRanking> receivedMessages)
 102         {
 103                 assertRankingEqualsRankingFromLastMessage(PETER, getLastMessageFor(PETER, receivedMessages));
 104                 assertRankingEqualsRankingFromLastMessage(KLAUS, getLastMessageFor(KLAUS, receivedMessages));
 105         }
 106
 107         private static void assertRankingEqualsRankingFromLastMessage(TestUser user, Ranking ranking)
 108         {
 109                 TestRanking testRanking = TestRanking.of(testEntriesOf(ranking.getEntries()));
 110                 assertRankingEqualsRankingFromLastMessage(user, testRanking);
 111         }
 112
 113         private static TestEntry[] testEntriesOf(Entry... entries)
 114         {
 115                 return Arrays
 116                                 .stream(entries)
 117                                 .map(entry -> TestEntry.of(
 118                                                 entry.getKey(),
 119                                                 entry.getCounter() == null
 120                                                                 ? -1l
 121                                                                 : entry.getCounter()))
 122                                 .toArray(size -> new TestEntry[size]);
 123         }
 124
 125         private static void assertRankingEqualsRankingFromLastMessage(TestUser user, TestRanking ranking)
 126         {
 127                 assertThat(ranking).isEqualTo(getLastMessageFor(user));
 128         }
 129
 130         private static TestRanking getLastMessageFor(TestUser user)
 131         {
 132                 return getLastMessageFor(user, expectedMessages());
 133         }
 134
 135         private static TestRanking getLastMessageFor(TestUser user, MultiValueMap<TestUser, TestRanking> messagesForUsers)
 136         {
 137                 return messagesForUsers
 138                                 .get(user)
 139                                 .stream()
 140                                 .reduce(null, (left, right) -> right);
 141         }
 142
 143         private static KeyValue<TestUser, TestRanking>[] EXPECTED_MESSAGES = new KeyValue[]
 144         {
 145                         KeyValue.pair( // 0
 146                                         PETER,
 147                                         TestRanking.of(
 148                                                         TestEntry.of("Hallo", 1l))),
 149                         KeyValue.pair( // 1
 150                                         KLAUS,
 151                                         TestRanking.of(
 152                                                         TestEntry.of("Müsch", 1l))),
 153                         KeyValue.pair( // 2
 154                                         PETER,
 155                                         TestRanking.of(
 156                                                         TestEntry.of("Hallo", 1l),
 157                                                         TestEntry.of("Welt", 1l))),
 158                         KeyValue.pair( // 3
 159                                         KLAUS,
 160                                         TestRanking.of(
 161                                                         TestEntry.of("Müsch", 2l))),
 162                         KeyValue.pair( // 4
 163                                         KLAUS,
 164                                         TestRanking.of(
 165                                                         TestEntry.of("Müsch", 2l),
 166                                                         TestEntry.of("s", 1l))),
 167                         KeyValue.pair( // 5
 168                                         PETER,
 169                                         TestRanking.of(
 170                                                         TestEntry.of("Hallo", 1l),
 171                                                         TestEntry.of("Welt", 1l),
 172                                                         TestEntry.of("Boäh", 1l))),
 173                         KeyValue.pair( // 6
 174                                         PETER,
 175                                         TestRanking.of(
 176                                                         TestEntry.of("Welt", 2l),
 177                                                         TestEntry.of("Hallo", 1l),
 178                                                         TestEntry.of("Boäh", 1l))),
 179                         KeyValue.pair( // 7
 180                                         PETER,
 181                                         TestRanking.of(
 182                                                         TestEntry.of("Welt", 2l),
 183                                                         TestEntry.of("Boäh", 2l),
 184                                                         TestEntry.of("Hallo", 1l))),
 185                         KeyValue.pair( // 8
 186                                         KLAUS,
 187                                         TestRanking.of(
 188                                                         TestEntry.of("Müsch", 2l),
 189                                                         TestEntry.of("s", 2l))),
 190                         KeyValue.pair( // 9
 191                                         PETER,
 192                                         TestRanking.of(
 193                                                         TestEntry.of("Boäh", 3l),
 194                                                         TestEntry.of("Welt", 2l),
 195                                                         TestEntry.of("Hallo", 1l))),
 196                         KeyValue.pair( // 10
 197                                         KLAUS,
 198                                         TestRanking.of(
 199                                                         TestEntry.of("s", 3l),
 200                                                         TestEntry.of("Müsch", 2l))),
 201         };
 202
 203         private static MultiValueMap<TestUser, TestRanking> expectedMessages()
 204         {
 205                 MultiValueMap<TestUser, TestRanking> expectedMessages = new LinkedMultiValueMap<>();
 206                 Stream
 207                                 .of(EXPECTED_MESSAGES)
 208                                 .forEach(keyValue -> expectedMessages.add(keyValue.key, keyValue.value));
 209                 return expectedMessages;
 210         }
 211 }