-package de.juplo.kafka.wordcount.counter;
-
-import de.juplo.kafka.wordcount.splitter.TestInputUser;
-import de.juplo.kafka.wordcount.splitter.TestInputWord;
-import de.juplo.kafka.wordcount.top10.TestOutputWord;
-import de.juplo.kafka.wordcount.top10.TestOutputWordCounter;
-import org.apache.kafka.streams.KeyValue;
-import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
-import org.springframework.util.LinkedMultiValueMap;
-import org.springframework.util.MultiValueMap;
-
-import java.util.stream.Stream;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-
-class TestData
-{
- static final String PETER = "peter";
- static final String KLAUS = "klaus";
-
- static final String WORD_HALLO = "Hallo";
- static final String WORD_MÜSCH = "Müsch";
- static final String WORD_WELT = "Welt";
- static final String WORD_S = "s";
- static final String WORD_BOÄH = "Boäh";
-
- static final TestOutputWord PETER_HALLO = TestOutputWord.of(PETER, WORD_HALLO);
- static final TestOutputWord PETER_WELT = TestOutputWord.of(PETER, WORD_WELT);
- static final TestOutputWord PETER_BOÄH = TestOutputWord.of(PETER, WORD_BOÄH);
- static final TestOutputWord KLAUS_MÜSCH = TestOutputWord.of(KLAUS, WORD_MÜSCH);
- static final TestOutputWord KLAUS_S = TestOutputWord.of(KLAUS, WORD_S);
-
- private static final KeyValue<TestInputUser, TestInputWord>[] INPUT_MESSAGES = new KeyValue[]
- {
- KeyValue.pair(
- TestInputUser.of(PETER),
- TestInputWord.of(PETER, WORD_HALLO)),
- KeyValue.pair(
- TestInputUser.of(KLAUS),
- TestInputWord.of(KLAUS, WORD_MÜSCH)),
- KeyValue.pair(
- TestInputUser.of(PETER),
- TestInputWord.of(PETER, WORD_WELT)),
- KeyValue.pair(
- TestInputUser.of(KLAUS),
- TestInputWord.of(KLAUS, WORD_MÜSCH)),
- KeyValue.pair(
- TestInputUser.of(KLAUS),
- TestInputWord.of(KLAUS, WORD_S)),
- KeyValue.pair(
- TestInputUser.of(PETER),
- TestInputWord.of(PETER, WORD_BOÄH)),
- KeyValue.pair(
- TestInputUser.of(PETER),
- TestInputWord.of(PETER, WORD_WELT)),
- KeyValue.pair(
- TestInputUser.of(PETER),
- TestInputWord.of(PETER, WORD_BOÄH)),
- KeyValue.pair(
- TestInputUser.of(KLAUS),
- TestInputWord.of(KLAUS, WORD_S)),
- KeyValue.pair(
- TestInputUser.of(PETER),
- TestInputWord.of(PETER, WORD_BOÄH)),
- KeyValue.pair(
- TestInputUser.of(KLAUS),
- TestInputWord.of(KLAUS, WORD_S)),
- };
-
- static Stream<KeyValue<TestInputUser, TestInputWord>> getInputMessages()
- {
- return Stream.of(TestData.INPUT_MESSAGES);
- }
-
- static void assertExpectedMessages(MultiValueMap<TestOutputWord, TestOutputWordCounter> receivedMessages)
- {
- expectedMessages().forEach(
- (word, counter) ->
- assertThat(receivedMessages.get(word))
- .containsExactlyElementsOf(counter));
- }
-
- static void assertExpectedNumberOfMessagesForWord(MultiValueMap<TestOutputWord, TestOutputWordCounter> receivedMessages)
- {
- assertThat(countMessagesForWord(PETER_HALLO, receivedMessages));
- assertThat(countMessagesForWord(PETER_WELT, receivedMessages));
- assertThat(countMessagesForWord(PETER_BOÄH, receivedMessages));
- assertThat(countMessagesForWord(KLAUS_MÜSCH, receivedMessages));
- assertThat(countMessagesForWord(KLAUS_S, receivedMessages));
- }
-
- private static int countMessagesForWord(TestOutputWord word, MultiValueMap<TestOutputWord, TestOutputWordCounter> messagesForUsers)
- {
- return messagesForUsers.get(word) == null
- ? 0
- : messagesForUsers.get(word).size();
- }
-
- static void assertExpectedState(ReadOnlyKeyValueStore<Word, Long> store)
- {
- assertWordCountEqualsWordCountFromLastMessage(PETER_HALLO, store.get(wordOf(PETER_HALLO)));
- assertWordCountEqualsWordCountFromLastMessage(PETER_WELT, store.get(wordOf(PETER_WELT)));
- assertWordCountEqualsWordCountFromLastMessage(PETER_BOÄH, store.get(wordOf(PETER_BOÄH)));
- assertWordCountEqualsWordCountFromLastMessage(KLAUS_MÜSCH, store.get(wordOf(KLAUS_MÜSCH)));
- assertWordCountEqualsWordCountFromLastMessage(KLAUS_S, store.get(wordOf(KLAUS_S)));
- }
-
- private static Word wordOf(TestOutputWord testOutputWord)
- {
- Word word = new Word();
-
- word.setUser(testOutputWord.getUser());
- word.setWord(testOutputWord.getWord());
-
- return word;
- }
-
- static void assertExpectedLastMessagesForWord(MultiValueMap<TestOutputWord, TestOutputWordCounter> receivedMessages)
- {
- assertWordCountEqualsWordCountFromLastMessage(PETER_HALLO, getLastMessageFor(PETER_HALLO, receivedMessages));
- assertWordCountEqualsWordCountFromLastMessage(PETER_WELT, getLastMessageFor(PETER_WELT, receivedMessages));
- assertWordCountEqualsWordCountFromLastMessage(PETER_BOÄH, getLastMessageFor(PETER_BOÄH, receivedMessages));
- assertWordCountEqualsWordCountFromLastMessage(KLAUS_MÜSCH, getLastMessageFor(KLAUS_MÜSCH, receivedMessages));
- assertWordCountEqualsWordCountFromLastMessage(KLAUS_S, getLastMessageFor(KLAUS_S, receivedMessages));
- }
-
- private static void assertWordCountEqualsWordCountFromLastMessage(
- TestOutputWord word,
- Long counter)
- {
- TestOutputWordCounter testOutputWordCounter = TestOutputWordCounter.of(
- word.getUser(),
- word.getWord(),
- counter);
- assertWordCountEqualsWordCountFromLastMessage(word, testOutputWordCounter);
- }
-
- private static void assertWordCountEqualsWordCountFromLastMessage(
- TestOutputWord word,
- TestOutputWordCounter counter)
- {
- assertThat(counter).isEqualTo(getLastMessageFor(word));
- }
-
- private static TestOutputWordCounter getLastMessageFor(TestOutputWord word)
- {
- return getLastMessageFor(word, expectedMessages());
- }
-
- private static TestOutputWordCounter getLastMessageFor(
- TestOutputWord user,
- MultiValueMap<TestOutputWord, TestOutputWordCounter> messagesForWord)
- {
- return messagesForWord
- .get(user)
- .stream()
- .reduce(null, (left, right) -> right);
- }
-
- private static final KeyValue<TestOutputWord, TestOutputWordCounter>[] EXPECTED_MESSAGES = new KeyValue[]
- {
- KeyValue.pair(
- PETER_HALLO,
- TestOutputWordCounter.of(PETER, WORD_HALLO,1)),
- KeyValue.pair(
- KLAUS_MÜSCH,
- TestOutputWordCounter.of(KLAUS, WORD_MÜSCH,1)),
- KeyValue.pair(
- PETER_WELT,
- TestOutputWordCounter.of(PETER, WORD_WELT,1)),
- KeyValue.pair(
- KLAUS_MÜSCH,
- TestOutputWordCounter.of(KLAUS, WORD_MÜSCH,2)),
- KeyValue.pair(
- KLAUS_S,
- TestOutputWordCounter.of(KLAUS, WORD_S,1)),
- KeyValue.pair(
- PETER_BOÄH,
- TestOutputWordCounter.of(PETER, WORD_BOÄH,1)),
- KeyValue.pair(
- PETER_WELT,
- TestOutputWordCounter.of(PETER, WORD_WELT,2)),
- KeyValue.pair(
- PETER_BOÄH,
- TestOutputWordCounter.of(PETER, WORD_BOÄH,2)),
- KeyValue.pair(
- KLAUS_S,
- TestOutputWordCounter.of(KLAUS, WORD_S,2)),
- KeyValue.pair(
- PETER_BOÄH,
- TestOutputWordCounter.of(PETER, WORD_BOÄH,3)),
- KeyValue.pair(
- KLAUS_S,
- TestOutputWordCounter.of(KLAUS, WORD_S,3)),
- };
-
- static MultiValueMap<TestOutputWord, TestOutputWordCounter> expectedMessages()
- {
- MultiValueMap<TestOutputWord, TestOutputWordCounter> expectedMessages = new LinkedMultiValueMap<>();
- Stream
- .of(EXPECTED_MESSAGES)
- .forEach(keyValue -> expectedMessages.add(keyValue.key, keyValue.value));
- return expectedMessages;
- }
-}