+ @DisplayName("Posted messages are accepted and sent to Kafka")
+ void userMessagesAreAcceptedAndSentToKafka()
+ {
+ MultiValueMap<String, TestRecording> recordings = new LinkedMultiValueMap<>();
+
+ Stream
+ .of(
+ new TestRecording("päter", "Hall° Wält?¢*&%€!"),
+ new TestRecording("päter", "Hallo Welt!"),
+ new TestRecording("klühs", "Müsch gäb's auch!"),
+ new TestRecording("päter", "Boäh, echt! ß mal nä Nümmäh!"))
+ .forEach(recording ->
+ {
+ sendRedording(recording.getUser(), recording.getSentence());
+ recordings.add(recording.getUser(), recording);
+ });
+
+
+ await("Received expected messages")
+ .atMost(Duration.ofSeconds(5))
+ .untilAsserted(() -> recordings.forEach((user, userRecordings) ->
+ assertThat(consumer.receivedFor(user)).containsExactlyElementsOf(userRecordings)));
+ }
+
+ void sendRedording(String user, String sentence)
+ {
+ try
+ {
+ MvcResult result = mockMvc
+ .perform(post("/{user}", user)
+ .contentType(MediaType.TEXT_PLAIN)
+ .content(sentence))
+ .andReturn();
+
+ mockMvc.perform(asyncDispatch(result))
+ .andDo(print())
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.username").value(user))
+ .andExpect(jsonPath("$.sentence").value(sentence));
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+
+
+ static class Consumer
+ {
+ private final MultiValueMap<String, TestRecording> received = new LinkedMultiValueMap<>();
+
+ @KafkaListener(groupId = "TEST", topics = TOPIC_OUT)
+ public synchronized void receive(
+ @Header(KafkaHeaders.RECEIVED_KEY) String user,
+ @Payload TestRecording recording)
+ {
+ log.debug("Received message: {}={}", user, recording);
+ received.add(user, recording);
+ }
+
+ synchronized List<TestRecording> receivedFor(String user)
+ {
+ List<TestRecording> recordings = received.get(user);
+ return recordings == null
+ ? List.of()
+ : received.get(user).stream().toList();
+ }
+ }
+
+ @TestConfiguration
+ static class Configuration