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