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()));
67 .takeUntil(message -> !running)
68 .doOnComplete(() -> log.info("TestWriter {} is done", user))
70 .runOn(Schedulers.parallel())
74 private Mono<MessageTo> sendMessage(
75 ChatRoomInfoTo chatRoom,
81 "/{chatRoomId}/{username}/{serial}",
85 .contentType(MediaType.TEXT_PLAIN)
86 .accept(MediaType.APPLICATION_JSON)
88 .exchangeToMono(response ->
90 if (response.statusCode().equals(HttpStatus.OK))
92 return response.bodyToMono(MessageTo.class);
96 return response.createError();
102 private final WebClient webClient;
103 private final ChatRoomInfoTo chatRoom;
106 final List<MessageTo> sentMessages = new LinkedList<>();
108 volatile boolean running = true;
111 TestWriter(Integer port, ChatRoomInfoTo chatRoom, String username)
113 webClient = WebClient.create("http://localhost:" + port);
114 this.chatRoom = chatRoom;
115 user = new User(username);