juplo.de Git - demos/kafka/chat/blob - src/test/java/de/juplo/kafka/chat/backend/HandoverIT.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.testcontainers.junit.jupiter.Testcontainers;
   9 import reactor.util.retry.Retry;
  10
  11 import java.time.Duration;
  12 import java.util.stream.IntStream;
  13
  14
  15 @Testcontainers
  16 @Slf4j
  17 public abstract class AbstractHandoverIT
  18 {
  19   private final AbstractContainerTemplates containerTemplates;
  20
  21
  22   @Test
  23   void test() throws InterruptedException
  24   {
  25     ChatRoomInfoTo chatRoom = createChatRoom("bar").block();
  26     User user = new User("nerd");
  27     IntStream
  28         .rangeClosed(1,100)
  29         .mapToObj(i ->sendMessage(chatRoom, user, "Message #" + i))
  30         .map(result -> result
  31             .map(MessageTo::toString)
  32             .retryWhen(Retry.fixedDelay(10, Duration.ofSeconds(1)))
  33             .block())
  34         .forEach(result -> log.info("{}", result));
  35
  36     receiveMessages(chatRoom)
  37         .take(100)
  38         .doOnNext(message -> log.info("message: {}", message))
  39         .then()
  40         .block();
  41   }
  42
  43   @BeforeEach
  44   void setUp() throws Exception
  45   {
  46     containerTemplates.setUp();
  47   }
  48 }