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