X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fwordcount%2Fcounter%2FTestData.java;h=43e1919b585c446a45832f081cecb275a0dd40ea;hb=35943a021ac4b25fad29861ecc083b6974ea2732;hp=dc3bcecfd98663b9f8d5dc0200804aaf39e3f5db;hpb=4072a93fd94a847c86ea422d74b17b0913b7de2a;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 dc3bcec..43e1919 100644 --- a/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java +++ b/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java @@ -1,51 +1,56 @@ package de.juplo.kafka.wordcount.counter; +import org.apache.kafka.streams.KeyValue; + import java.util.List; +import java.util.Map; +import java.util.Properties; import java.util.function.BiConsumer; +import java.util.stream.Collectors; import static org.assertj.core.api.Assertions.assertThat; class TestData { - static void writeInputData(BiConsumer consumer) + static void writeInputData(BiConsumer consumer) { consumer.accept( "peter", - "Hallo"); + Word.of("peter","Hallo")); consumer.accept( "klaus", - "Müsch"); + Word.of("klaus","Müsch")); consumer.accept( "peter", - "Welt"); + Word.of("peter","Welt")); consumer.accept( "klaus", - "Müsch"); + Word.of("klaus","Müsch")); consumer.accept( "klaus", - "s"); + Word.of("klaus","s")); consumer.accept( "peter", - "Boäh"); + Word.of("peter","Boäh")); consumer.accept( "peter", - "Welt"); + Word.of("peter","Welt")); consumer.accept( "peter", - "Boäh"); + Word.of("peter","Boäh")); consumer.accept( "klaus", - "s"); + Word.of("klaus","s")); consumer.accept( "peter", - "Boäh"); + Word.of("peter","Boäh")); consumer.accept( "klaus", - "s"); + Word.of("klaus","s")); } - static void assertExpectedResult(List receivedMessages) + static void assertExpectedResult(List> receivedMessages) { assertThat(receivedMessages).hasSize(11); assertThat(receivedMessages).containsSubsequence( @@ -66,40 +71,52 @@ class TestData expectedMessages[9]); // Boäh } - static Message[] expectedMessages = + static KeyValue[] expectedMessages = new KeyValue[] { - Message.of( - "{\"username\":\"peter\",\"word\":\"Hallo\"}", - "1"), - Message.of( - "{\"username\":\"klaus\",\"word\":\"Müsch\"}", - "1"), - Message.of( - "{\"username\":\"peter\",\"word\":\"Welt\"}", - "1"), - Message.of( - "{\"username\":\"klaus\",\"word\":\"Müsch\"}", - "2"), - Message.of( - "{\"username\":\"klaus\",\"word\":\"s\"}", - "1"), - Message.of( - "{\"username\":\"peter\",\"word\":\"Boäh\"}", - "1"), - Message.of( - "{\"username\":\"peter\",\"word\":\"Welt\"}", - "2"), - Message.of( - "{\"username\":\"peter\",\"word\":\"Boäh\"}", - "2"), - Message.of( - "{\"username\":\"klaus\",\"word\":\"s\"}", - "2"), - Message.of( - "{\"username\":\"peter\",\"word\":\"Boäh\"}", - "3"), - Message.of( - "{\"username\":\"klaus\",\"word\":\"s\"}", - "3"), + KeyValue.pair( + Word.of("peter","Hallo"), + WordCounter.of("peter","Hallo",1)), + KeyValue.pair( + Word.of("klaus","Müsch"), + WordCounter.of("klaus","Müsch",1)), + KeyValue.pair( + Word.of("peter","Welt"), + WordCounter.of("peter","Welt",1)), + KeyValue.pair( + Word.of("klaus","Müsch"), + WordCounter.of("klaus","Müsch",2)), + KeyValue.pair( + Word.of("klaus","s"), + WordCounter.of("klaus","s",1)), + KeyValue.pair( + Word.of("peter","Boäh"), + WordCounter.of("peter","Boäh",1)), + KeyValue.pair( + Word.of("peter","Welt"), + WordCounter.of("peter","Welt",2)), + KeyValue.pair( + Word.of("peter","Boäh"), + WordCounter.of("peter","Boäh",2)), + KeyValue.pair( + Word.of("klaus","s"), + WordCounter.of("klaus","s",2)), + KeyValue.pair( + Word.of("peter","Boäh"), + WordCounter.of("peter","Boäh",3)), + KeyValue.pair( + Word.of("klaus","s"), + WordCounter.of("klaus","s",3)), }; + + static Map convertToMap(Properties properties) + { + return properties + .entrySet() + .stream() + .collect( + Collectors.toMap( + entry -> (String)entry.getKey(), + entry -> entry.getValue() + )); + } }