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
  14
  15 @Slf4j
  16 public class TestClient
  17 {
  18   public void run()
  19   {
  20     for (int i = 0; i < 100; i++)
  21     {
  22       String message = "Message #" + i;
  23       for (ChatRoomInfoTo chatRoom : chatRooms)
  24       {
  25         sendMessage(chatRoom, message)
  26             .retryWhen(Retry.fixedDelay(10, Duration.ofSeconds(1)))
  27             .map(MessageTo::toString)
  28             .subscribe(result -> log.info(
  29                 "{} sent message \"{}\" to {}",
  30                 user,
  31                 message,
  32                 chatRoom));
  33       }
  34     }
  35   }
  36
  37   private Mono<MessageTo> sendMessage(
  38       ChatRoomInfoTo chatRoom,
  39       String message)
  40   {
  41     return webClient
  42         .put()
  43         .uri(
  44             "/{chatRoomId}/{username}/{serial}",
  45             chatRoom.getId(),
  46             user.getName(),
  47             user.nextSerial())
  48         .contentType(MediaType.TEXT_PLAIN)
  49         .accept(MediaType.APPLICATION_JSON)
  50         .bodyValue(message)
  51         .exchangeToMono(response ->
  52         {
  53           if (response.statusCode().equals(HttpStatus.OK))
  54           {
  55             return response.bodyToMono(MessageTo.class);
  56           }
  57           else
  58           {
  59             return response.createError();
  60           }
  61         });
  62   }
  63
  64
  65   private final WebClient webClient;
  66   private final ChatRoomInfoTo[] chatRooms;
  67   private final User user;
  68
  69
  70   TestClient(Integer port, ChatRoomInfoTo[] chatRooms, String username)
  71   {
  72     webClient = WebClient.create("http://localhost:" + port);
  73     this.chatRooms = chatRooms;
  74     user = new User(username);
  75   }
  76 }