From: Kai Moritz Date: Mon, 13 Feb 2023 18:00:53 +0000 (+0100) Subject: counter: 1.1.6 - Refactored the test-data into a separated class X-Git-Tag: counter-1.1.6 X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=0a7e20af38fe1502c7c4328b2b6024cfe9e7b019;p=demos%2Fkafka%2Fwordcount counter: 1.1.6 - Refactored the test-data into a separated class --- diff --git a/pom.xml b/pom.xml index 9231396..09e7aec 100644 --- a/pom.xml +++ b/pom.xml @@ -10,7 +10,7 @@ de.juplo.kafka.wordcount counter - 1.1.5 + 1.1.6 Wordcount-Counter Word-counting stream-processor of the multi-user wordcount-example diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java b/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java index fd92000..75f0988 100644 --- a/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java +++ b/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java @@ -1,9 +1,7 @@ package de.juplo.kafka.wordcount.counter; -import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import lombok.RequiredArgsConstructor; -import lombok.Value; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier; @@ -18,13 +16,12 @@ import org.springframework.context.annotation.Primary; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.test.context.EmbeddedKafka; -import org.springframework.util.LinkedMultiValueMap; -import org.springframework.util.MultiValueMap; import java.time.Duration; +import java.util.LinkedList; +import java.util.List; import static de.juplo.kafka.wordcount.counter.CounterApplicationIT.*; -import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.*; @@ -60,95 +57,36 @@ public class CounterApplicationIT @Test - void testSendMessage() throws Exception + void testSendMessage() { - kafkaTemplate.send(TOPIC_IN, "peter", "Hallo"); - kafkaTemplate.send(TOPIC_IN, "klaus", "Müsch"); - kafkaTemplate.send(TOPIC_IN, "peter", "Welt"); - kafkaTemplate.send(TOPIC_IN, "klaus", "Müsch"); - kafkaTemplate.send(TOPIC_IN, "klaus", "s"); - kafkaTemplate.send(TOPIC_IN, "peter", "Boäh"); - kafkaTemplate.send(TOPIC_IN, "peter", "Welt"); - kafkaTemplate.send(TOPIC_IN, "peter", "Boäh"); - kafkaTemplate.send(TOPIC_IN, "klaus", "s"); - kafkaTemplate.send(TOPIC_IN, "peter", "Boäh"); - kafkaTemplate.send(TOPIC_IN, "klaus", "s"); - - Message peter1 = Message.of( - "{\"username\":\"peter\",\"word\":\"Hallo\"}", - "1"); - Message peter2 = Message.of( - "{\"username\":\"peter\",\"word\":\"Welt\"}", - "1"); - Message peter3 = Message.of( - "{\"username\":\"peter\",\"word\":\"Boäh\"}", - "1"); - Message peter4 = Message.of( - "{\"username\":\"peter\",\"word\":\"Welt\"}", - "2"); - Message peter5 = Message.of( - "{\"username\":\"peter\",\"word\":\"Boäh\"}", - "2"); - Message peter6 = Message.of( - "{\"username\":\"peter\",\"word\":\"Boäh\"}", - "3"); - - Message klaus1 = Message.of( - "{\"username\":\"klaus\",\"word\":\"Müsch\"}", - "1"); - Message klaus2 = Message.of( - "{\"username\":\"klaus\",\"word\":\"Müsch\"}", - "2"); - Message klaus3 = Message.of( - "{\"username\":\"klaus\",\"word\":\"s\"}", - "1"); - Message klaus4 = Message.of( - "{\"username\":\"klaus\",\"word\":\"s\"}", - "2"); - Message klaus5 = Message.of( - "{\"username\":\"klaus\",\"word\":\"s\"}", - "3"); + TestData.writeInputData((key, value) -> kafkaTemplate.send(TOPIC_IN, key, value)); await("Expexted converted data") .atMost(Duration.ofSeconds(10)) - .untilAsserted(() -> - { - assertThat(consumer.received).hasSize(2); - assertThat(consumer.received.get("klaus")).containsExactly(klaus1, klaus2, klaus3, klaus4, klaus5); - assertThat(consumer.received.get("peter")).containsExactly(peter1, peter2, peter3, peter4, peter5, peter6); - }); + .untilAsserted(() -> TestData.assertExpectedResult(consumer.received, mapper)); } @RequiredArgsConstructor static class Consumer { - private final MultiValueMap received = new LinkedMultiValueMap<>(); - private final ObjectMapper mapper; + private final List received = new LinkedList<>(); @KafkaListener(groupId = "TEST", topics = TOPIC_OUT) - public void receive(ConsumerRecord record) throws JsonProcessingException + public void receive(ConsumerRecord record) { log.debug("Received message: {}", record); - Key key = mapper.readValue(record.key(), Key.class); - received.add(key.getUsername(), Message.of(record.key(),record.value())); + received.add(Message.of(record.key(),record.value())); } } - @Value(staticConstructor = "of") - static class Message - { - String key; - String value; - } - @TestConfiguration static class Configuration { @Bean - Consumer consumer(ObjectMapper mapper) + Consumer consumer() { - return new Consumer(mapper); + return new Consumer(); } @Primary diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/Message.java b/src/test/java/de/juplo/kafka/wordcount/counter/Message.java new file mode 100644 index 0000000..8ed7cc5 --- /dev/null +++ b/src/test/java/de/juplo/kafka/wordcount/counter/Message.java @@ -0,0 +1,11 @@ +package de.juplo.kafka.wordcount.counter; + +import lombok.Value; + + +@Value(staticConstructor = "of") +public class Message +{ + String key; + String value; +} diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java b/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java new file mode 100644 index 0000000..4b5482b --- /dev/null +++ b/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java @@ -0,0 +1,109 @@ +package de.juplo.kafka.wordcount.counter; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.springframework.util.LinkedMultiValueMap; +import org.springframework.util.MultiValueMap; + +import java.util.List; +import java.util.function.BiConsumer; + +import static org.assertj.core.api.Assertions.assertThat; + + +class TestData +{ + static void writeInputData(BiConsumer consumer) + { + consumer.accept( + "peter", + "Hallo"); + consumer.accept( + "klaus", + "Müsch"); + consumer.accept( + "peter", + "Welt"); + consumer.accept( + "klaus", + "Müsch"); + consumer.accept( + "klaus", + "s"); + consumer.accept( + "peter", + "Boäh"); + consumer.accept( + "peter", + "Welt"); + consumer.accept( + "peter", + "Boäh"); + consumer.accept( + "klaus", + "s"); + consumer.accept( + "peter", + "Boäh"); + consumer.accept( + "klaus", + "s"); + } + + static void assertExpectedResult(List receivedMessages, ObjectMapper mapper) + { + MultiValueMap received = new LinkedMultiValueMap<>(); + + receivedMessages.forEach(message -> + { + try + { + Key key = mapper.readValue(message.getKey(), Key.class); + received.add(key.getUsername(), message); + } + catch (JsonProcessingException e) + { + throw new RuntimeException(e); + } + }); + + assertThat(received).hasSize(2); + assertThat(received.get("klaus")).containsExactly(klaus1, klaus2, klaus3, klaus4, klaus5); + assertThat(received.get("peter")).containsExactly(peter1, peter2, peter3, peter4, peter5, peter6); + } + + static Message peter1 = Message.of( + "{\"username\":\"peter\",\"word\":\"Hallo\"}", + "1"); + static Message peter2 = Message.of( + "{\"username\":\"peter\",\"word\":\"Welt\"}", + "1"); + static Message peter3 = Message.of( + "{\"username\":\"peter\",\"word\":\"Boäh\"}", + "1"); + static Message peter4 = Message.of( + "{\"username\":\"peter\",\"word\":\"Welt\"}", + "2"); + static Message peter5 = Message.of( + "{\"username\":\"peter\",\"word\":\"Boäh\"}", + "2"); + static Message peter6 = Message.of( + "{\"username\":\"peter\",\"word\":\"Boäh\"}", + "3"); + + static Message klaus1 = Message.of( + "{\"username\":\"klaus\",\"word\":\"Müsch\"}", + "1"); + static Message klaus2 = Message.of( + "{\"username\":\"klaus\",\"word\":\"Müsch\"}", + "2"); + static Message klaus3 = Message.of( + "{\"username\":\"klaus\",\"word\":\"s\"}", + "1"); + static Message klaus4 = Message.of( + "{\"username\":\"klaus\",\"word\":\"s\"}", + "2"); + static Message klaus5 = Message.of( + "{\"username\":\"klaus\",\"word\":\"s\"}", + "3"); +}