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;
{
consumer.accept(
"peter",
- "Hallo");
+ "{\"user\":\"peter\",\"word\":\"Hallo\"}");
consumer.accept(
"klaus",
- "Müsch");
+ "{\"user\":\"klaus\",\"word\":\"Müsch\"}");
consumer.accept(
"peter",
- "Welt");
+ "{\"user\":\"peter\",\"word\":\"Welt\"}");
consumer.accept(
"klaus",
- "Müsch");
+ "{\"user\":\"klaus\",\"word\":\"Müsch\"}");
consumer.accept(
"klaus",
- "s");
+ "{\"user\":\"klaus\",\"word\":\"s\"}");
consumer.accept(
"peter",
- "Boäh");
+ "{\"user\":\"peter\",\"word\":\"Boäh\"}");
consumer.accept(
"peter",
- "Welt");
+ "{\"user\":\"peter\",\"word\":\"Welt\"}");
consumer.accept(
"peter",
- "Boäh");
+ "{\"user\":\"peter\",\"word\":\"Boäh\"}");
consumer.accept(
"klaus",
- "s");
+ "{\"user\":\"klaus\",\"word\":\"s\"}");
consumer.accept(
"peter",
- "Boäh");
+ "{\"user\":\"peter\",\"word\":\"Boäh\"}");
consumer.accept(
"klaus",
- "s");
+ "{\"user\":\"klaus\",\"word\":\"s\"}");
}
- static void assertExpectedResult(List<Message> receivedMessages, ObjectMapper mapper)
+ static void assertExpectedResult(List<Message> receivedMessages)
{
- MultiValueMap<String, Message> 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);
+ assertThat(receivedMessages).hasSize(11);
+ assertThat(receivedMessages).containsSubsequence(
+ expectedMessages[0]); // Hallo
+ assertThat(receivedMessages).containsSubsequence(
+ expectedMessages[1],
+ expectedMessages[3]); // Müsch
+ assertThat(receivedMessages).containsSubsequence(
+ expectedMessages[2],
+ expectedMessages[6]);
+ assertThat(receivedMessages).containsSubsequence(
+ expectedMessages[4],
+ expectedMessages[8],
+ expectedMessages[10]); // s
+ assertThat(receivedMessages).containsSubsequence(
+ expectedMessages[5],
+ expectedMessages[7],
+ expectedMessages[9]); // Boäh
}
- 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");
+ static Message[] expectedMessages =
+ {
+ Message.of(
+ "{\"user\":\"peter\",\"word\":\"Hallo\"}",
+ "{\"user\":\"peter\",\"word\":\"Hallo\",\"count\":1}"),
+ Message.of(
+ "{\"user\":\"klaus\",\"word\":\"Müsch\"}",
+ "{\"user\":\"klaus\",\"word\":\"Müsch\",\"count\":1}"),
+ Message.of(
+ "{\"user\":\"peter\",\"word\":\"Welt\"}",
+ "{\"user\":\"peter\",\"word\":\"Welt\",\"count\":1}"),
+ Message.of(
+ "{\"user\":\"klaus\",\"word\":\"Müsch\"}",
+ "{\"user\":\"klaus\",\"word\":\"Müsch\",\"count\":2}"),
+ Message.of(
+ "{\"user\":\"klaus\",\"word\":\"s\"}",
+ "{\"user\":\"klaus\",\"word\":\"s\",\"count\":1}"),
+ Message.of(
+ "{\"user\":\"peter\",\"word\":\"Boäh\"}",
+ "{\"user\":\"peter\",\"word\":\"Boäh\",\"count\":1}"),
+ Message.of(
+ "{\"user\":\"peter\",\"word\":\"Welt\"}",
+ "{\"user\":\"peter\",\"word\":\"Welt\",\"count\":2}"),
+ Message.of(
+ "{\"user\":\"peter\",\"word\":\"Boäh\"}",
+ "{\"user\":\"peter\",\"word\":\"Boäh\",\"count\":2}"),
+ Message.of(
+ "{\"user\":\"klaus\",\"word\":\"s\"}",
+ "{\"user\":\"klaus\",\"word\":\"s\",\"count\":2}"),
+ Message.of(
+ "{\"user\":\"peter\",\"word\":\"Boäh\"}",
+ "{\"user\":\"peter\",\"word\":\"Boäh\",\"count\":3}"),
+ Message.of(
+ "{\"user\":\"klaus\",\"word\":\"s\"}",
+ "{\"user\":\"klaus\",\"word\":\"s\",\"count\":3}"),
+ };
}