juplo.de Git - demos/kafka/chat/blob - src/test/java/de/juplo/kafka/chat/backend/TestListener.java

   1 package de.juplo.kafka.chat.backend;
   2
   3 import com.fasterxml.jackson.databind.ObjectMapper;
   4 import com.fasterxml.jackson.databind.SerializationFeature;
   5 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
   6 import de.juplo.kafka.chat.backend.api.ChatRoomInfoTo;
   7 import de.juplo.kafka.chat.backend.api.MessageTo;
   8 import lombok.extern.slf4j.Slf4j;
   9 import org.springframework.core.ParameterizedTypeReference;
  10 import org.springframework.http.MediaType;
  11 import org.springframework.http.codec.ServerSentEvent;
  12 import org.springframework.web.reactive.function.client.WebClient;
  13 import reactor.core.publisher.Flux;
  14 import reactor.core.publisher.Mono;
  15 import reactor.core.scheduler.Schedulers;
  16
  17 import java.util.*;
  18
  19
  20 @Slf4j
  21 public class TestListener
  22 {
  23   static final ParameterizedTypeReference<ServerSentEvent<String>> SSE_TYPE = new ParameterizedTypeReference<>() {};
  24
  25
  26   public Mono<Void> run()
  27   {
  28     return Flux
  29         .fromArray(chatRooms)
  30         .flatMap(chatRoom ->
  31         {
  32           log.info("Requesting messages from chat-room {}", chatRoom);
  33           List<MessageTo> list = new LinkedList<>();
  34           receivedMessages.put(chatRoom.getId(), list);
  35           return receiveMessages(chatRoom)
  36               .flatMap(sse ->
  37               {
  38                 try
  39                 {
  40                   return Mono.just(objectMapper.readValue(sse.data(), MessageTo.class));
  41                 }
  42                 catch (Exception e)
  43                 {
  44                   return Mono.error(e);
  45                 }
  46               })
  47               .doOnNext(message ->
  48               {
  49                 list.add(message);
  50                 log.info(
  51                     "Received a message from chat-room {}: {}",
  52                     chatRoom.getName(),
  53                     message);
  54               })
  55               .take(10);
  56         })
  57         .take(100)
  58         .takeUntil(message -> !running)
  59         .doOnComplete(() -> log.info("TestListener is done"))
  60         .parallel(chatRooms.length)
  61         .runOn(Schedulers.parallel())
  62         .then();
  63   }
  64
  65   Flux<ServerSentEvent<String>> receiveMessages(ChatRoomInfoTo chatRoom)
  66   {
  67     return webClient
  68         .get()
  69         .uri(
  70             "/{chatRoomId}/listen",
  71             chatRoom.getId())
  72         .accept(MediaType.TEXT_EVENT_STREAM)
  73         .retrieve()
  74         .bodyToFlux(SSE_TYPE);
  75   }
  76
  77
  78   private final WebClient webClient;
  79   private final ChatRoomInfoTo[] chatRooms;
  80   private final ObjectMapper objectMapper;
  81
  82   final Map<UUID, List<MessageTo>> receivedMessages = new HashMap<>();
  83
  84   volatile boolean running = true;
  85
  86
  87   TestListener(Integer port, ChatRoomInfoTo[] chatRooms)
  88   {
  89     webClient = WebClient.create("http://localhost:" + port);
  90     this.chatRooms = chatRooms;
  91     objectMapper = new ObjectMapper();
  92     objectMapper.registerModule(new JavaTimeModule());
  93     objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
  94   }
  95 }