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.core.ParameterizedTypeReference;
   7 import org.springframework.http.HttpStatus;
   8 import org.springframework.http.MediaType;
   9 import org.springframework.http.codec.ServerSentEvent;
  10 import org.springframework.web.reactive.function.client.WebClient;
  11 import reactor.core.publisher.Flux;
  12 import reactor.core.publisher.Mono;
  13
  14
  15 @Slf4j
  16 public class TestClient
  17 {
  18   static final ParameterizedTypeReference<ServerSentEvent<String>> SSE_TYPE = new ParameterizedTypeReference<>() {};
  19
  20
  21   Mono<ChatRoomInfoTo> createChatRoom(String name)
  22   {
  23     return webClient
  24         .post()
  25         .uri("/create")
  26         .contentType(MediaType.TEXT_PLAIN)
  27         .bodyValue(name)
  28         .accept(MediaType.APPLICATION_JSON)
  29         .exchangeToMono(response ->
  30         {
  31           if (response.statusCode().equals(HttpStatus.OK))
  32           {
  33             return response.bodyToMono(ChatRoomInfoTo.class);
  34           }
  35           else
  36           {
  37             return response.createError();
  38           }
  39         });
  40   }
  41
  42   Mono<MessageTo> sendMessage(
  43       ChatRoomInfoTo chatRoom,
  44       User user,
  45       String message)
  46   {
  47     return webClient
  48         .put()
  49         .uri(
  50             "/{chatRoomId}/{username}/{serial}",
  51             chatRoom.getId(),
  52             user.getName(),
  53             user.nextSerial())
  54         .contentType(MediaType.TEXT_PLAIN)
  55         .accept(MediaType.APPLICATION_JSON)
  56         .bodyValue(message)
  57         .exchangeToMono(response ->
  58         {
  59           if (response.statusCode().equals(HttpStatus.OK))
  60           {
  61             return response.bodyToMono(MessageTo.class);
  62           }
  63           else
  64           {
  65             return response.createError();
  66           }
  67         });
  68   }
  69
  70   Flux<ServerSentEvent<String>> receiveMessages(ChatRoomInfoTo chatRoom)
  71   {
  72     return webClient
  73         .get()
  74         .uri(
  75             "/{chatRoomId}/listen",
  76             chatRoom.getId())
  77         .accept(MediaType.TEXT_EVENT_STREAM)
  78         .retrieve()
  79         .bodyToFlux(SSE_TYPE);
  80   }
  81
  82
  83   private final WebClient webClient;
  84
  85
  86   TestClient(Integer port)
  87   {
  88     webClient = WebClient.create("http://localhost:" + port);
  89   }
  90 }