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.core.scheduler.Schedulers;
13 import reactor.util.retry.Retry;
15 import java.nio.charset.Charset;
16 import java.time.Duration;
17 import java.util.Iterator;
18 import java.util.LinkedList;
19 import java.util.List;
20 import java.util.concurrent.ThreadLocalRandom;
24 public class TestWriter
26 public Mono<Void> run()
29 .fromIterable((Iterable<Integer>) () -> new Iterator<>()
34 public boolean hasNext()
45 .map(i -> "Message #" + i)
46 .flatMap(message -> sendMessage(chatRoom, message)
47 .delayElement(Duration.ofMillis(ThreadLocalRandom.current().nextLong(500, 1500)))
48 .retryWhen(Retry.fixedDelay(10, Duration.ofSeconds(1))))
51 sentMessages.add(message);
53 "{} sent a message to {}: {}",
58 .doOnError(throwable ->
60 WebClientResponseException e = (WebClientResponseException)throwable.getCause();
62 "{} failed sending a message: {}",
64 e.getResponseBodyAsString(Charset.defaultCharset()));
66 .takeUntil(message -> !running)
67 .doOnComplete(() -> log.info("TestWriter {} is done", user))
69 .runOn(Schedulers.parallel())
73 private Mono<MessageTo> sendMessage(
74 ChatRoomInfoTo chatRoom,
80 "/{chatRoomId}/{username}/{serial}",
84 .contentType(MediaType.TEXT_PLAIN)
85 .accept(MediaType.APPLICATION_JSON)
87 .exchangeToMono(response ->
89 if (response.statusCode().equals(HttpStatus.OK))
91 return response.bodyToMono(MessageTo.class);
95 return response.createError();
101 private final WebClient webClient;
102 private final ChatRoomInfoTo chatRoom;
105 final List<MessageTo> sentMessages = new LinkedList<>();
107 volatile boolean running = true;
110 TestWriter(Integer port, ChatRoomInfoTo chatRoom, String username)
112 webClient = WebClient.create("http://localhost:" + port);
113 this.chatRoom = chatRoom;
114 user = new User(username);