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