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.junit.jupiter.api.BeforeEach;
   7 import org.junit.jupiter.api.Disabled;
   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.util.List;
  18 import java.util.concurrent.CompletableFuture;
  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
  31
  32   AbstractHandoverIT(AbstractHandoverITContainers containers)
  33   {
  34     this.containers = containers;
  35   }
  36
  37
  38   @Disabled
  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     CompletableFuture<Void>[] testWriterFutures = new CompletableFuture[NUM_CLIENTS];
  51     TestWriter[] testWriters = new TestWriter[NUM_CLIENTS];
  52     for (int i = 0; i < NUM_CLIENTS; i++)
  53     {
  54       TestWriter testWriter = new TestWriter(
  55           port,
  56           chatRooms[i % NUM_CHATROOMS],
  57           "user-" + i);
  58       testWriters[i] = testWriter;
  59       testWriterFutures[i] = testWriter
  60           .run()
  61           .toFuture();
  62     }
  63
  64     TestListener testListener = new TestListener(port, chatRooms);
  65     testListener
  66         .run()
  67         .subscribe(message -> log.info(
  68             "Received message: {}",
  69             message));
  70
  71     log.info("Sleeping for 3 seconds...");
  72     Thread.sleep(3000);
  73
  74     log.info("Starting Backend 2...");
  75     containers.backend2.start();
  76     log.info("Backend 2 started");
  77
  78     log.info("Sleeping for 3 seconds...");
  79     Thread.sleep(3000);
  80
  81     log.info("Starting Backend 3...");
  82     containers.backend3.start();
  83     log.info("Backend 3 started");
  84
  85     log.info("Sleeping for 3 seconds...");
  86     Thread.sleep(3000);
  87
  88     log.info("Shutting down writers...");
  89     for (int i = 0; i < NUM_CLIENTS; i++)
  90     {
  91       log.info("Shutting down writer {}", i);
  92       testWriters[i].running = false;
  93       testWriterFutures[i].join();
  94       log.info("Joined TestWriter {}", testWriters[i].user);
  95     }
  96
  97     // Yield the work, so that the last messages can be received
  98     Thread.sleep(500);
  99
 100     for (int i = 0; i < NUM_CLIENTS; i++)
 101     {
 102       TestWriter testWriter = testWriters[i];
 103       ChatRoomInfoTo chatRoom = testWriter.chatRoom;
 104       List<MessageTo> receivedMessages = testListener.receivedMessages.get(chatRoom.getId());
 105
 106       Assertions.assertThat(receivedMessages
 107           .stream()
 108           .filter(message -> message.getUser().equals(testWriter.user.getName()))
 109           ).containsExactlyElementsOf(testWriter.sentMessages);
 110     }
 111   }
 112
 113   Mono<ChatRoomInfoTo> createChatRoom(String name)
 114   {
 115     return webClient
 116         .post()
 117         .uri("/create")
 118         .contentType(MediaType.TEXT_PLAIN)
 119         .bodyValue(name)
 120         .accept(MediaType.APPLICATION_JSON)
 121         .exchangeToMono(response ->
 122         {
 123           if (response.statusCode().equals(HttpStatus.OK))
 124           {
 125             return response.bodyToMono(ChatRoomInfoTo.class);
 126           }
 127           else
 128           {
 129             return response.createError();
 130           }
 131         });
 132   }
 133
 134
 135   WebClient webClient;
 136
 137   @BeforeEach
 138   void setUp() throws Exception
 139   {
 140     containers.setUp();
 141
 142     Integer port = containers.haproxy.getMappedPort(8400);
 143     webClient = WebClient.create("http://localhost:" + port);
 144   }
 145 }