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
  20
  21 @Testcontainers
  22 @Slf4j
  23 public abstract class AbstractHandoverIT
  24 {
  25   static final int NUM_CHATROOMS = 23;
  26   static final int NUM_CLIENTS = 17;
  27
  28
  29   private final AbstractHandoverITContainers containers;
  30   private final ExecutorService executorService = Executors.newFixedThreadPool(NUM_CLIENTS + 1);
  31
  32
  33   AbstractHandoverIT(AbstractHandoverITContainers containers)
  34   {
  35     this.containers = containers;
  36   }
  37
  38
  39   @Test
  40   void test() throws InterruptedException
  41   {
  42     ChatRoomInfoTo[] chatRooms = Flux
  43         .range(0, NUM_CHATROOMS)
  44         .flatMap(i -> createChatRoom("room-" + i))
  45         .toStream()
  46         .toArray(size -> new ChatRoomInfoTo[size]);
  47
  48     int port = containers.haproxy.getMappedPort(8400);
  49
  50     TestWriter[] testWriters = Flux
  51         .range(0, NUM_CLIENTS)
  52         .map(i -> new TestWriter(
  53             port,
  54             chatRooms[i % NUM_CHATROOMS],
  55             "user-" + Integer.toString(i)))
  56         .doOnNext(testClient -> executorService.execute(testClient))
  57         .toStream()
  58         .toArray(size -> new TestWriter[size]);
  59
  60     TestListener testListener = new TestListener(port, chatRooms);
  61     executorService.execute(testListener);
  62
  63     Thread.sleep(10000);
  64
  65     Arrays
  66         .stream(testWriters)
  67         .forEach(testClient -> testClient.running = false);
  68     testListener.running = false;
  69   }
  70
  71   Mono<ChatRoomInfoTo> createChatRoom(String name)
  72   {
  73     return webClient
  74         .post()
  75         .uri("/create")
  76         .contentType(MediaType.TEXT_PLAIN)
  77         .bodyValue(name)
  78         .accept(MediaType.APPLICATION_JSON)
  79         .exchangeToMono(response ->
  80         {
  81           if (response.statusCode().equals(HttpStatus.OK))
  82           {
  83             return response.bodyToMono(ChatRoomInfoTo.class);
  84           }
  85           else
  86           {
  87             return response.createError();
  88           }
  89         });
  90   }
  91
  92
  93   WebClient webClient;
  94
  95   @BeforeEach
  96   void setUp() throws Exception
  97   {
  98     containers.setUp();
  99
 100     Integer port = containers.haproxy.getMappedPort(8400);
 101     webClient = WebClient.create("http://localhost:" + port);
 102   }
 103 }