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 de.juplo.kafka.chat.backend.api.MessageTo;
   5 import lombok.extern.slf4j.Slf4j;
   6 import org.awaitility.Awaitility;
   7 import org.junit.jupiter.api.BeforeEach;
   8 import org.junit.jupiter.api.Test;
   9 import org.springframework.http.HttpStatus;
  10 import org.springframework.http.MediaType;
  11 import org.springframework.web.reactive.function.client.WebClient;
  12 import org.testcontainers.junit.jupiter.Testcontainers;
  13 import pl.rzrz.assertj.reactor.Assertions;
  14 import reactor.core.publisher.Flux;
  15 import reactor.core.publisher.Mono;
  16
  17 import java.time.Duration;
  18 import java.util.List;
  19 import java.util.concurrent.CompletableFuture;
  20
  21
  22 @Testcontainers
  23 @Slf4j
  24 public abstract class AbstractHandoverIT
  25 {
  26   static final int NUM_CHATROOMS = 23;
  27   static final int NUM_CLIENTS = 17;
  28
  29
  30   private final AbstractHandoverITContainers containers;
  31
  32
  33   AbstractHandoverIT(AbstractHandoverITContainers containers)
  34   {
  35     this.containers = containers;
  36   }
  37
  38
  39   @Test
  40   void test() throws InterruptedException
  41   {
  42     log.info("Starting backend-1...");
  43     containers.startBackend(containers.backend1, new TestWriter[0]);
  44     log.info("backend-1 started!");
  45
  46     ChatRoomInfoTo[] chatRooms = Flux
  47         .range(0, NUM_CHATROOMS)
  48         .flatMap(i -> createChatRoom("room-" + i))
  49         .toStream()
  50         .toArray(size -> new ChatRoomInfoTo[size]);
  51
  52     int port = containers.haproxy.getMappedPort(8400);
  53
  54     CompletableFuture<Void>[] testWriterFutures = new CompletableFuture[NUM_CLIENTS];
  55     TestWriter[] testWriters = new TestWriter[NUM_CLIENTS];
  56     for (int i = 0; i < NUM_CLIENTS; i++)
  57     {
  58       TestWriter testWriter = new TestWriter(
  59           port,
  60           chatRooms[i % NUM_CHATROOMS],
  61           "user-" + i);
  62       testWriters[i] = testWriter;
  63       testWriterFutures[i] = testWriter
  64           .run()
  65           .toFuture();
  66     }
  67
  68     TestListener testListener = new TestListener(port, chatRooms);
  69     testListener
  70         .run()
  71         .subscribe(message -> log.info(
  72             "Received message: {}",
  73             message));
  74
  75     log.info("Starting backend-2...");
  76     containers.startBackend(containers.backend2, testWriters);
  77     log.info("backend-2 started!");
  78
  79     for (int i = 0; i < NUM_CLIENTS; i++)
  80     {
  81       testWriters[i].running = false;
  82       testWriterFutures[i].join();
  83       log.info("Joined TestWriter {}", testWriters[i].user);
  84     }
  85
  86     Awaitility
  87         .await()
  88         .atMost(Duration.ofSeconds(30))
  89         .untilAsserted(() -> assertAllSentMessagesReceived(testWriters, testListener));
  90   }
  91
  92   private void assertAllSentMessagesReceived(
  93       TestWriter[] testWriters,
  94       TestListener testListener)
  95   {
  96     for (int i = 0; i < NUM_CLIENTS; i++)
  97     {
  98       TestWriter testWriter = testWriters[i];
  99       ChatRoomInfoTo chatRoom = testWriter.chatRoom;
 100       List<MessageTo> receivedMessages = testListener.receivedMessages.get(chatRoom.getId());
 101
 102       Assertions.assertThat(receivedMessages
 103           .stream()
 104           .filter(message -> message.getUser().equals(testWriter.user.getName()))
 105           ).containsExactlyElementsOf(testWriter.sentMessages);
 106     }
 107   }
 108
 109   Mono<ChatRoomInfoTo> createChatRoom(String name)
 110   {
 111     return webClient
 112         .post()
 113         .uri("/create")
 114         .contentType(MediaType.TEXT_PLAIN)
 115         .bodyValue(name)
 116         .accept(MediaType.APPLICATION_JSON)
 117         .exchangeToMono(response ->
 118         {
 119           if (response.statusCode().equals(HttpStatus.OK))
 120           {
 121             return response.bodyToMono(ChatRoomInfoTo.class);
 122           }
 123           else
 124           {
 125             return response.createError();
 126           }
 127         });
 128   }
 129
 130
 131   WebClient webClient;
 132
 133   @BeforeEach
 134   void setUp() throws Exception
 135   {
 136     containers.setUp();
 137
 138     Integer port = containers.haproxy.getMappedPort(8400);
 139     webClient = WebClient.create("http://localhost:" + port);
 140   }
 141 }