import java.util.stream.Stream;
+import static de.juplo.kafka.wordcount.counter.CounterStreamProcessor.TYPE;
import static org.assertj.core.api.Assertions.assertThat;
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);
+ static final TestOutputWord PETER_HALLO = TestOutputWord.of(TYPE, PETER, WORD_HALLO);
+ static final TestOutputWord PETER_WELT = TestOutputWord.of(TYPE, PETER, WORD_WELT);
+ static final TestOutputWord PETER_BOÄH = TestOutputWord.of(TYPE, PETER, WORD_BOÄH);
+ static final TestOutputWord KLAUS_MÜSCH = TestOutputWord.of(TYPE, KLAUS, WORD_MÜSCH);
+ static final TestOutputWord KLAUS_S = TestOutputWord.of(TYPE, KLAUS, WORD_S);
private static final KeyValue<TestInputUser, TestInputWord>[] INPUT_MESSAGES = new KeyValue[]
{
- new KeyValue<>(
+ KeyValue.pair(
TestInputUser.of(PETER),
TestInputWord.of(PETER, WORD_HALLO)),
- new KeyValue<>(
+ KeyValue.pair(
TestInputUser.of(KLAUS),
TestInputWord.of(KLAUS, WORD_MÜSCH)),
- new KeyValue<>(
+ KeyValue.pair(
TestInputUser.of(PETER),
TestInputWord.of(PETER, WORD_WELT)),
- new KeyValue<>(
+ KeyValue.pair(
TestInputUser.of(KLAUS),
TestInputWord.of(KLAUS, WORD_MÜSCH)),
- new KeyValue<>(
+ KeyValue.pair(
TestInputUser.of(KLAUS),
TestInputWord.of(KLAUS, WORD_S)),
- new KeyValue<>(
+ KeyValue.pair(
TestInputUser.of(PETER),
TestInputWord.of(PETER, WORD_BOÄH)),
- new KeyValue<>(
+ KeyValue.pair(
TestInputUser.of(PETER),
TestInputWord.of(PETER, WORD_WELT)),
- new KeyValue<>(
+ KeyValue.pair(
TestInputUser.of(PETER),
TestInputWord.of(PETER, WORD_BOÄH)),
- new KeyValue<>(
+ KeyValue.pair(
TestInputUser.of(KLAUS),
TestInputWord.of(KLAUS, WORD_S)),
- new KeyValue<>(
+ KeyValue.pair(
TestInputUser.of(PETER),
TestInputWord.of(PETER, WORD_BOÄH)),
- new KeyValue<>(
+ KeyValue.pair(
TestInputUser.of(KLAUS),
TestInputWord.of(KLAUS, WORD_S)),
};
private static int countMessagesForWord(TestOutputWord word, MultiValueMap<TestOutputWord, TestOutputWordCounter> messagesForUsers)
{
- return messagesForUsers.get(word).size();
+ return messagesForUsers.get(word) == null
+ ? 0
+ : messagesForUsers.get(word).size();
}
static void assertExpectedState(ReadOnlyKeyValueStore<Word, Long> store)
private static Word wordOf(TestOutputWord testOutputWord)
{
- Word word = new Word();
-
- word.setUser(testOutputWord.getUser());
- word.setWord(testOutputWord.getWord());
-
- return word;
+ return Word.of(
+ testOutputWord.getChannel(),
+ testOutputWord.getKey());
}
static void assertExpectedLastMessagesForWord(MultiValueMap<TestOutputWord, TestOutputWordCounter> receivedMessages)
Long counter)
{
TestOutputWordCounter testOutputWordCounter = TestOutputWordCounter.of(
- word.getUser(),
- word.getWord(),
+ word.getChannel(),
+ word.getKey(),
counter);
assertWordCountEqualsWordCountFromLastMessage(word, testOutputWordCounter);
}