juplo.de Git - demos/kafka/chat/blob - src/test/java/de/juplo/kafka/chat/backend/TestClient.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.springframework.http.HttpStatus;
   7 import org.springframework.http.MediaType;
   8 import org.springframework.web.reactive.function.client.WebClient;
   9 import reactor.core.publisher.Mono;
  10 import reactor.util.retry.Retry;
  11
  12 import java.time.Duration;
  13 import java.util.stream.IntStream;
  14
  15
  16 @Slf4j
  17 public class TestClient
  18 {
  19   public void run()
  20   {
  21     IntStream
  22         .rangeClosed(1,100)
  23         .mapToObj(i ->sendMessage(chatRoom, "Message #" + i))
  24         .map(result -> result
  25             .map(MessageTo::toString)
  26             .retryWhen(Retry.fixedDelay(10, Duration.ofSeconds(1)))
  27             .block())
  28         .forEach(result -> log.info("{}", result));
  29   }
  30
  31   private Mono<MessageTo> sendMessage(
  32       ChatRoomInfoTo chatRoom,
  33       String message)
  34   {
  35     return webClient
  36         .put()
  37         .uri(
  38             "/{chatRoomId}/{username}/{serial}",
  39             chatRoom.getId(),
  40             user.getName(),
  41             user.nextSerial())
  42         .contentType(MediaType.TEXT_PLAIN)
  43         .accept(MediaType.APPLICATION_JSON)
  44         .bodyValue(message)
  45         .exchangeToMono(response ->
  46         {
  47           if (response.statusCode().equals(HttpStatus.OK))
  48           {
  49             return response.bodyToMono(MessageTo.class);
  50           }
  51           else
  52           {
  53             return response.createError();
  54           }
  55         });
  56   }
  57
  58
  59   private final WebClient webClient;
  60   private final ChatRoomInfoTo chatRoom;
  61   private final User user;
  62
  63
  64   TestClient(Integer port, ChatRoomInfoTo chatRoom, String username)
  65   {
  66     webClient = WebClient.create("http://localhost:" + port);
  67     this.chatRoom = chatRoom;
  68     user = new User(username);
  69   }
  70 }