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

   1 package de.juplo.kafka.chat.backend;
   2
   3 import de.juplo.kafka.chat.backend.api.ChatRoomInfoTo;
   4 import lombok.extern.slf4j.Slf4j;
   5 import org.junit.jupiter.api.BeforeEach;
   6 import org.junit.jupiter.api.Test;
   7 import org.springframework.core.ParameterizedTypeReference;
   8 import org.springframework.http.HttpStatus;
   9 import org.springframework.http.MediaType;
  10 import org.springframework.http.codec.ServerSentEvent;
  11 import org.springframework.web.reactive.function.client.WebClient;
  12 import org.testcontainers.junit.jupiter.Testcontainers;
  13 import reactor.core.publisher.Flux;
  14 import reactor.core.publisher.Mono;
  15
  16 import java.util.concurrent.ExecutorService;
  17 import java.util.concurrent.Executors;
  18 import java.util.stream.IntStream;
  19
  20
  21 @Testcontainers
  22 @Slf4j
  23 public abstract class AbstractHandoverIT
  24 {
  25   static final ParameterizedTypeReference<ServerSentEvent<String>> SSE_TYPE = new ParameterizedTypeReference<>() {};
  26   static final int NUM_CHATROOMS = 23;
  27   static final int NUM_CLIENTS = 17;
  28
  29
  30   private final AbstractHandoverITContainers containers;
  31   private final ExecutorService executorService = Executors.newFixedThreadPool(NUM_CLIENTS);
  32
  33
  34   AbstractHandoverIT(AbstractHandoverITContainers containers)
  35   {
  36     this.containers = containers;
  37   }
  38
  39
  40   @Test
  41   void test() throws InterruptedException
  42   {
  43     ChatRoomInfoTo[] chatRooms = Flux
  44         .range(0, NUM_CHATROOMS)
  45         .flatMap(i -> createChatRoom("#" + i))
  46         .toStream()
  47         .toArray(size -> new ChatRoomInfoTo[size]);
  48
  49     TestClient[] testClients = Flux
  50         .fromStream(IntStream.range(0, NUM_CLIENTS).mapToObj(i -> Integer.toString(i)))
  51         .map(i -> new TestClient(
  52             containers.haproxy.getMappedPort(8400),
  53             chatRooms,
  54             i))
  55         .doOnNext(testClient -> executorService.execute(testClient))
  56         .toStream()
  57         .toArray(size -> new TestClient[size]);
  58
  59     Thread.sleep(2000);
  60
  61     Flux
  62         .fromArray(chatRooms)
  63         .flatMap(chatRoom ->receiveMessages(chatRoom).take(100))
  64         .doOnNext(message -> log.info("message: {}", message))
  65         .then()
  66         .block();
  67   }
  68
  69   Mono<ChatRoomInfoTo> createChatRoom(String name)
  70   {
  71     return webClient
  72         .post()
  73         .uri("/create")
  74         .contentType(MediaType.TEXT_PLAIN)
  75         .bodyValue(name)
  76         .accept(MediaType.APPLICATION_JSON)
  77         .exchangeToMono(response ->
  78         {
  79           if (response.statusCode().equals(HttpStatus.OK))
  80           {
  81             return response.bodyToMono(ChatRoomInfoTo.class);
  82           }
  83           else
  84           {
  85             return response.createError();
  86           }
  87         });
  88   }
  89
  90   Flux<ServerSentEvent<String>> receiveMessages(ChatRoomInfoTo chatRoom)
  91   {
  92     return webClient
  93         .get()
  94         .uri(
  95             "/{chatRoomId}/listen",
  96             chatRoom.getId())
  97         .accept(MediaType.TEXT_EVENT_STREAM)
  98         .retrieve()
  99         .bodyToFlux(SSE_TYPE);
 100   }
 101
 102
 103   WebClient webClient;
 104
 105   @BeforeEach
 106   void setUp() throws Exception
 107   {
 108     containers.setUp();
 109
 110     Integer port = containers.haproxy.getMappedPort(8400);
 111     webClient = WebClient.create("http://localhost:" + port);
 112   }
 113 }