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