X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fwordcount%2Fcounter%2FTestData.java;h=714e5850a8a4c6e8da383d352a01fbff1af82be3;hb=ed364e3fb4436eeb181b6eac9ff4650d3d91f21c;hp=6110f8084461be745c3815e34c63d96d66ac0a7c;hpb=7474f0e1fc7028e60ef9526fec86bfb8824c6517;p=demos%2Fkafka%2Fwordcount diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java b/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java index 6110f80..714e585 100644 --- a/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java +++ b/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java @@ -1,15 +1,13 @@ package de.juplo.kafka.wordcount.counter; -import org.apache.kafka.common.header.Header; -import org.apache.kafka.common.header.Headers; +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.springframework.util.LinkedMultiValueMap; import org.springframework.util.MultiValueMap; -import java.util.Map; -import java.util.Properties; -import java.util.function.BiConsumer; -import java.util.stream.Collectors; +import java.util.function.Consumer; import java.util.stream.Stream; import static org.assertj.core.api.Assertions.assertThat; @@ -17,104 +15,119 @@ import static org.assertj.core.api.Assertions.assertThat; class TestData { - static void injectInputMessages(BiConsumer consumer) + 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[] INPUT_MESSAGES = new KeyValue[] { - Stream - .of(INPUT_MESSAGES) - .forEach(word -> consumer.accept(word.getUser(), word)); + new KeyValue<>( + PETER, + TestInputWord.of(PETER, WORD_HALLO)), + new KeyValue<>( + KLAUS, + TestInputWord.of(KLAUS, WORD_MÜSCH)), + new KeyValue<>( + PETER, + TestInputWord.of(PETER, WORD_WELT)), + new KeyValue<>( + KLAUS, + TestInputWord.of(KLAUS, WORD_MÜSCH)), + new KeyValue<>( + KLAUS, + TestInputWord.of(KLAUS, WORD_S)), + new KeyValue<>( + PETER, + TestInputWord.of(PETER, WORD_BOÄH)), + new KeyValue<>( + PETER, + TestInputWord.of(PETER, WORD_WELT)), + new KeyValue<>( + PETER, + TestInputWord.of(PETER, WORD_BOÄH)), + new KeyValue<>( + KLAUS, + TestInputWord.of(KLAUS, WORD_S)), + new KeyValue<>( + PETER, + TestInputWord.of(PETER, WORD_BOÄH)), + new KeyValue<>( + KLAUS, + TestInputWord.of(KLAUS, WORD_S)), + }; + + static Stream> getInputMessages() + { + return Stream.of(TestData.INPUT_MESSAGES); } - static final Word[] INPUT_MESSAGES = new Word[] + static Consumer> expectedMessagesAssertion() { - Word.of("peter","Hallo"), - Word.of("klaus","Müsch"), - Word.of("peter","Welt"), - Word.of("klaus","Müsch"), - Word.of("klaus","s"), - Word.of("peter","Boäh"), - Word.of("peter","Welt"), - Word.of("peter","Boäh"), - Word.of("klaus","s"), - Word.of("peter","Boäh"), - Word.of("klaus","s"), - }; + return receivedMessages -> assertExpectedMessages(receivedMessages); + } - static void assertExpectedMessages(MultiValueMap receivedMessages) + static void assertExpectedMessages(MultiValueMap receivedMessages) { - expectedMessages.forEach( + expectedMessages().forEach( (word, counter) -> assertThat(receivedMessages.get(word)) .containsExactlyElementsOf(counter)); } - static final KeyValue[] EXPECTED_MESSAGES = new KeyValue[] + private static final KeyValue[] EXPECTED_MESSAGES = new KeyValue[] { KeyValue.pair( - Word.of("peter","Hallo"), - WordCounter.of("peter","Hallo",1)), + PETER_HALLO, + TestOutputWordCounter.of(PETER, WORD_HALLO,1)), KeyValue.pair( - Word.of("klaus","Müsch"), - WordCounter.of("klaus","Müsch",1)), + KLAUS_MÜSCH, + TestOutputWordCounter.of(KLAUS, WORD_MÜSCH,1)), KeyValue.pair( - Word.of("peter","Welt"), - WordCounter.of("peter","Welt",1)), + PETER_WELT, + TestOutputWordCounter.of(PETER, WORD_WELT,1)), KeyValue.pair( - Word.of("klaus","Müsch"), - WordCounter.of("klaus","Müsch",2)), + KLAUS_MÜSCH, + TestOutputWordCounter.of(KLAUS, WORD_MÜSCH,2)), KeyValue.pair( - Word.of("klaus","s"), - WordCounter.of("klaus","s",1)), + KLAUS_S, + TestOutputWordCounter.of(KLAUS, WORD_S,1)), KeyValue.pair( - Word.of("peter","Boäh"), - WordCounter.of("peter","Boäh",1)), + PETER_BOÄH, + TestOutputWordCounter.of(PETER, WORD_BOÄH,1)), KeyValue.pair( - Word.of("peter","Welt"), - WordCounter.of("peter","Welt",2)), + PETER_WELT, + TestOutputWordCounter.of(PETER, WORD_WELT,2)), KeyValue.pair( - Word.of("peter","Boäh"), - WordCounter.of("peter","Boäh",2)), + PETER_BOÄH, + TestOutputWordCounter.of(PETER, WORD_BOÄH,2)), KeyValue.pair( - Word.of("klaus","s"), - WordCounter.of("klaus","s",2)), + KLAUS_S, + TestOutputWordCounter.of(KLAUS, WORD_S,2)), KeyValue.pair( - Word.of("peter","Boäh"), - WordCounter.of("peter","Boäh",3)), + PETER_BOÄH, + TestOutputWordCounter.of(PETER, WORD_BOÄH,3)), KeyValue.pair( - Word.of("klaus","s"), - WordCounter.of("klaus","s",3)), + KLAUS_S, + TestOutputWordCounter.of(KLAUS, WORD_S,3)), }; - static MultiValueMap expectedMessages; - static + static MultiValueMap expectedMessages() { - expectedMessages = new LinkedMultiValueMap<>(); + MultiValueMap expectedMessages = new LinkedMultiValueMap<>(); Stream .of(EXPECTED_MESSAGES) .forEach(keyValue -> expectedMessages.add(keyValue.key, keyValue.value)); - } - - static Map convertToMap(Properties properties) - { - return properties - .entrySet() - .stream() - .collect( - Collectors.toMap( - entry -> (String)entry.getKey(), - entry -> entry.getValue() - )); - } - - static String parseHeader(Headers headers, String key) - { - Header header = headers.lastHeader(key); - if (header == null) - { - return key + "=null"; - } - else - { - return key + "=" + new String(header.value()); - } + return expectedMessages; } }