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.http.HttpStatus;
   8 import org.springframework.http.MediaType;
   9 import org.springframework.web.reactive.function.client.WebClient;
  10 import org.testcontainers.junit.jupiter.Testcontainers;
  11 import reactor.core.publisher.Flux;
  12 import reactor.core.publisher.Mono;
  13
  14 import java.util.concurrent.CompletableFuture;
  15
  16
  17 @Testcontainers
  18 @Slf4j
  19 public abstract class AbstractHandoverIT
  20 {
  21   static final int NUM_CHATROOMS = 23;
  22   static final int NUM_CLIENTS = 17;
  23
  24
  25   private final AbstractHandoverITContainers containers;
  26
  27
  28   AbstractHandoverIT(AbstractHandoverITContainers containers)
  29   {
  30     this.containers = containers;
  31   }
  32
  33
  34   @Test
  35   void test() throws InterruptedException
  36   {
  37     ChatRoomInfoTo[] chatRooms = Flux
  38         .range(0, NUM_CHATROOMS)
  39         .flatMap(i -> createChatRoom("room-" + i))
  40         .toStream()
  41         .toArray(size -> new ChatRoomInfoTo[size]);
  42
  43     int port = containers.haproxy.getMappedPort(8400);
  44
  45     CompletableFuture<Void>[] testWriterFutures = new CompletableFuture[NUM_CLIENTS];
  46     TestWriter[] testWriters = new TestWriter[NUM_CLIENTS];
  47     for (int i = 0; i < NUM_CLIENTS; i++)
  48     {
  49       TestWriter testWriter = new TestWriter(
  50           port,
  51           chatRooms[i % NUM_CHATROOMS],
  52           "user-" + i);
  53       testWriters[i] = testWriter;
  54       testWriterFutures[i] = testWriter
  55           .run()
  56           .toFuture();
  57     }
  58
  59     TestListener testListener = new TestListener(port, chatRooms);
  60     CompletableFuture<Void> testListenerFuture = testListener
  61         .run()
  62         .toFuture();
  63
  64     log.info("Sleeping for 3 seconds...");
  65     Thread.sleep(3000);
  66
  67     for (int i = 0; i < NUM_CLIENTS; i++)
  68     {
  69       testWriters[i].running = false;
  70       testWriterFutures[i].join();
  71       log.info("Joined TestWriter {}", testWriters[i].user);
  72     }
  73
  74
  75     log.info("Sleeping for 3 seconds...");
  76     Thread.sleep(3000);
  77     log.info("Joining TestListener...");
  78     testListener.running = false;
  79     testListenerFuture.join();
  80     log.info("Joined TestListener");
  81   }
  82
  83   Mono<ChatRoomInfoTo> createChatRoom(String name)
  84   {
  85     return webClient
  86         .post()
  87         .uri("/create")
  88         .contentType(MediaType.TEXT_PLAIN)
  89         .bodyValue(name)
  90         .accept(MediaType.APPLICATION_JSON)
  91         .exchangeToMono(response ->
  92         {
  93           if (response.statusCode().equals(HttpStatus.OK))
  94           {
  95             return response.bodyToMono(ChatRoomInfoTo.class);
  96           }
  97           else
  98           {
  99             return response.createError();
 100           }
 101         });
 102   }
 103
 104
 105   WebClient webClient;
 106
 107   @BeforeEach
 108   void setUp() throws Exception
 109   {
 110     containers.setUp();
 111
 112     Integer port = containers.haproxy.getMappedPort(8400);
 113     webClient = WebClient.create("http://localhost:" + port);
 114   }
 115 }