package de.juplo.kafka.wordcount.recorder;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.context.TestConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.http.MediaType;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.messaging.handler.annotation.Payload;
+import org.springframework.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.MvcResult;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
-@SpringBootTest
+import java.time.Duration;
+import java.util.stream.Stream;
+
+import static de.juplo.kafka.wordcount.recorder.ApplicationTests.TOPIC_OUT;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.asyncDispatch;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
+import static org.springframework.test.web.servlet.result.MockMvcResultHandlers.print;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+
+@SpringBootTest(
+ properties = {
+ "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
+ "spring.kafka.consumer.auto-offset-reset=earliest",
+ "spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer",
+ "spring.kafka.consumer.properties.spring.json.value.default.type=de.juplo.kafka.wordcount.recorder.RecordingData",
+ "spring.kafka.consumer.properties.spring.json.trusted.packages=de.juplo.kafka.wordcount.recorder",
+ "logging.level.root=WARN",
+ "logging.level.de.juplo=DEBUG",
+ "juplo.wordcount.recorder.bootstrap-server=${spring.embedded.kafka.brokers}",
+ "juplo.wordcount.recorder.topic=" + TOPIC_OUT })
+@AutoConfigureMockMvc
+@EmbeddedKafka(topics = { TOPIC_OUT })
+@Slf4j
class ApplicationTests
{
+ static final String TOPIC_OUT = "out";
+
+ @Autowired
+ MockMvc mockMvc;
+ @Autowired
+ Consumer consumer;
+
+
@Test
- void contextLoads()
+ @DisplayName("Posted messages are accepted and sent to Kafka")
+ void userMessagesAreAcceptedAndSentToKafka()
+ {
+ MultiValueMap<String, RecordingData> recordings = new LinkedMultiValueMap<>();
+
+ Stream
+ .of(
+ new RecordingData("päter", "Hall° Wält?¢*&%€!"),
+ new RecordingData("päter", "Hallo Welt!"),
+ new RecordingData("klühs", "Müsch gäb's auch!"),
+ new RecordingData("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(10))
+ .untilAsserted(() -> recordings.forEach((user, userRecordings) ->
+ assertThat(consumer.received.get(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
+ {
+ final MultiValueMap<String, RecordingData> received = new LinkedMultiValueMap<>();
+
+ @KafkaListener(groupId = "TEST", topics = TOPIC_OUT)
+ public void receive(
+ @Header(KafkaHeaders.RECEIVED_KEY) String key,
+ @Payload RecordingData recording)
+ {
+ log.debug("Received message: {}={}", key, recording);
+ received.add(key, recording);
+ }
+ }
+
+ @TestConfiguration
+ static class Configuration
{
+ @Bean
+ Consumer consumer()
+ {
+ return new Consumer();
+ }
}
}