1 package de.juplo.kafka.chat.backend;
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 org.springframework.web.reactive.function.client.WebClientResponseException;
10 import reactor.core.publisher.Flux;
11 import reactor.core.publisher.Mono;
12 import reactor.util.retry.Retry;
14 import java.nio.charset.Charset;
15 import java.time.Duration;
16 import java.util.Iterator;
17 import java.util.concurrent.ThreadLocalRandom;
21 public class TestWriter implements Runnable
27 .fromIterable((Iterable<Integer>) () -> new Iterator<>()
32 public boolean hasNext()
43 .map(i -> "Message #" + i)
44 .flatMap(message -> sendMessage(chatRoom, message)
45 .delayElement(Duration.ofMillis(ThreadLocalRandom.current().nextLong(500, 1500)))
46 .retryWhen(Retry.fixedDelay(10, Duration.ofSeconds(1))))
47 .doOnNext(message -> log.info(
48 "{} sent a message to {}: {}",
52 .doOnError(throwable ->
54 WebClientResponseException e = (WebClientResponseException)throwable.getCause();
56 "{} failed sending a message: {}",
58 e.getResponseBodyAsString(Charset.defaultCharset()));
64 private Mono<MessageTo> sendMessage(
65 ChatRoomInfoTo chatRoom,
71 "/{chatRoomId}/{username}/{serial}",
75 .contentType(MediaType.TEXT_PLAIN)
76 .accept(MediaType.APPLICATION_JSON)
78 .exchangeToMono(response ->
80 if (response.statusCode().equals(HttpStatus.OK))
82 return response.bodyToMono(MessageTo.class);
86 return response.createError();
92 private final WebClient webClient;
93 private final ChatRoomInfoTo chatRoom;
94 private final User user;
96 volatile boolean running = true;
99 TestWriter(Integer port, ChatRoomInfoTo chatRoom, String username)
101 webClient = WebClient.create("http://localhost:" + port);
102 this.chatRoom = chatRoom;
103 user = new User(username);