X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fchat%2Fbackend%2FAbstractHandoverIT.java;h=7ab65f6f8faf62af15ae3188acc7b38d8c1a4da8;hb=319a01f87982b8cf00a0b77839b69d161f32d606;hp=f384f7e8afdf641aa20f2ae1def14db5a177d015;hpb=76e4e8f96472a4a298629e9fa61f18a2c2056010;p=demos%2Fkafka%2Fchat diff --git a/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java b/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java index f384f7e8..7ab65f6f 100644 --- a/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java +++ b/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java @@ -1,13 +1,7 @@ package de.juplo.kafka.chat.backend; import de.juplo.kafka.chat.backend.api.ChatRoomInfoTo; -import de.juplo.kafka.chat.backend.api.MessageTo; -import lombok.EqualsAndHashCode; -import lombok.Getter; -import lombok.Setter; -import lombok.ToString; import lombok.extern.slf4j.Slf4j; -import org.awaitility.Awaitility; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.springframework.core.ParameterizedTypeReference; @@ -15,20 +9,13 @@ import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; import org.springframework.http.codec.ServerSentEvent; import org.springframework.web.reactive.function.client.WebClient; -import org.testcontainers.containers.BindMode; -import org.testcontainers.containers.GenericContainer; -import org.testcontainers.containers.Network; -import org.testcontainers.containers.output.Slf4jLogConsumer; -import org.testcontainers.containers.wait.strategy.Wait; -import org.testcontainers.images.ImagePullPolicy; import org.testcontainers.junit.jupiter.Testcontainers; -import org.testcontainers.utility.DockerImageName; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import reactor.util.retry.Retry; -import java.io.IOException; -import java.time.Duration; +import java.util.Arrays; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.stream.IntStream; @@ -36,26 +23,49 @@ import java.util.stream.IntStream; @Slf4j public abstract class AbstractHandoverIT { - static final ImagePullPolicy NEVER_PULL = imageName -> false; static final ParameterizedTypeReference> SSE_TYPE = new ParameterizedTypeReference<>() {}; + static final int NUM_CHATROOMS = 23; + static final int NUM_CLIENTS = 17; + + + private final AbstractHandoverITContainers containers; + private final ExecutorService executorService = Executors.newFixedThreadPool(NUM_CLIENTS); + + + AbstractHandoverIT(AbstractHandoverITContainers containers) + { + this.containers = containers; + } @Test void test() throws InterruptedException { - ChatRoomInfoTo chatRoom = createChatRoom("bar").block(); - User user = new User("nerd"); - IntStream - .rangeClosed(1,100) - .mapToObj(i ->sendMessage(chatRoom, user, "Message #" + i)) - .map(result -> result - .map(MessageTo::toString) - .retryWhen(Retry.fixedDelay(10, Duration.ofSeconds(1))) - .block()) - .forEach(result -> log.info("{}", result)); - - receiveMessages(chatRoom) - .take(100) + ChatRoomInfoTo[] chatRooms = Flux + .range(0, NUM_CHATROOMS) + .flatMap(i -> createChatRoom("room-" + i)) + .toStream() + .toArray(size -> new ChatRoomInfoTo[size]); + + TestClient[] testClients = Flux + .fromStream(IntStream.range(0, NUM_CLIENTS).mapToObj(i -> "user-" + Integer.toString(i))) + .map(i -> new TestClient( + containers.haproxy.getMappedPort(8400), + chatRooms, + i)) + .doOnNext(testClient -> executorService.execute(testClient)) + .toStream() + .toArray(size -> new TestClient[size]); + + Thread.sleep(2000); + + Arrays + .stream(testClients) + .forEach(testClient -> testClient.running = false); + + Flux + .fromArray(chatRooms) + .flatMap(chatRoom ->receiveMessages(chatRoom).take(50)) .doOnNext(message -> log.info("message: {}", message)) .then() .block(); @@ -82,34 +92,6 @@ public abstract class AbstractHandoverIT }); } - Mono sendMessage( - ChatRoomInfoTo chatRoom, - User user, - String message) - { - return webClient - .put() - .uri( - "/{chatRoomId}/{username}/{serial}", - chatRoom.getId(), - user.getName(), - user.nextSerial()) - .contentType(MediaType.TEXT_PLAIN) - .accept(MediaType.APPLICATION_JSON) - .bodyValue(message) - .exchangeToMono(response -> - { - if (response.statusCode().equals(HttpStatus.OK)) - { - return response.bodyToMono(MessageTo.class); - } - else - { - return response.createError(); - } - }); - } - Flux> receiveMessages(ChatRoomInfoTo chatRoom) { return webClient @@ -125,150 +107,12 @@ public abstract class AbstractHandoverIT WebClient webClient; - - abstract void setUpExtra() throws IOException, InterruptedException; - @BeforeEach void setUp() throws Exception { - setUpExtra(); - haproxy.start(); - backend1.start(); - // backend2.start(); - // backend3.start(); + containers.setUp(); - Integer port = haproxy.getMappedPort(8400); + Integer port = containers.haproxy.getMappedPort(8400); webClient = WebClient.create("http://localhost:" + port); - - Awaitility - .await() - .atMost(Duration.ofMinutes(10)) - .until(() -> WebClient - .create("http://localhost:" + backend1.getMappedPort(8080)) - .get() - .uri("/actuator/health") - .exchangeToMono(response -> - { - if (response.statusCode().equals(HttpStatus.OK)) - { - return response - .bodyToMono(StatusTo.class) - .map(StatusTo::getStatus) - .map(status -> status.equalsIgnoreCase("UP")); - } - else - { - return Mono.just(false); - } - }) - .block()); - - haproxy - .getDockerClient() - .killContainerCmd(haproxy.getContainerId()) - .withSignal("HUP") - .exec(); - - - Awaitility - .await() - .atMost(Duration.ofMinutes(10)) - .until(() -> webClient - .get() - .uri("/actuator/health") - .exchangeToMono(response -> - { - if (response.statusCode().equals(HttpStatus.OK)) - { - return response - .bodyToMono(StatusTo.class) - .map(StatusTo::getStatus) - .map(status -> status.equalsIgnoreCase("UP")); - } - else - { - return Mono.just(false); - } - }) - .block()); - } - - Network network = Network.newNetwork(); - - GenericContainer haproxy = - new GenericContainer(DockerImageName.parse("haproxytech/haproxy-debian:2.8")) - .withNetwork(network) - .withNetworkAliases("haproxy") - .withClasspathResourceMapping( - "haproxy.cfg", - "/usr/local/etc/haproxy/haproxy.cfg", - BindMode.READ_ONLY) - .withClasspathResourceMapping( - "sharding.map", - "/usr/local/etc/haproxy/sharding.map", - BindMode.READ_WRITE) - .withExposedPorts(8400, 8401, 8404) - .withLogConsumer(new Slf4jLogConsumer(log, true).withPrefix("HAPROXY")); - - abstract String[] getCommandBackend1(); - GenericContainer backend1 = - new GenericContainer(DockerImageName.parse("juplo/chat-backend:0.0.1-SNAPSHOT")) - .withImagePullPolicy(NEVER_PULL) - .withNetwork(network) - .withNetworkAliases("backend-1") - .withCommand(getCommandBackend1()) - .withExposedPorts(8080) - .waitingFor(Wait.forLogMessage(".*Started\\ ChatBackendApplication.*\\n", 1)) - .withLogConsumer(new Slf4jLogConsumer(log, true).withPrefix("BACKEND-1")); - - abstract String[] getCommandBackend2(); - GenericContainer backend2 = - new GenericContainer(DockerImageName.parse("juplo/chat-backend:0.0.1-SNAPSHOT")) - .withImagePullPolicy(NEVER_PULL) - .withNetwork(network) - .withNetworkAliases("backend-2") - .withCommand(getCommandBackend2()) - .withExposedPorts(8080) - .waitingFor(Wait.forLogMessage(".*Started\\ ChatBackendApplication.*\\n", 1)) - .withLogConsumer(new Slf4jLogConsumer(log, true).withPrefix("BACKEND-2")); - - abstract String[] getCommandBackend3(); - GenericContainer backend3 = - new GenericContainer(DockerImageName.parse("juplo/chat-backend:0.0.1-SNAPSHOT")) - .withImagePullPolicy(NEVER_PULL) - .withNetwork(network) - .withNetworkAliases("backend-3") - .withCommand(getCommandBackend3()) - .withExposedPorts(8080) - .waitingFor(Wait.forLogMessage(".*Started\\ ChatBackendApplication.*\\n", 1)) - .withLogConsumer(new Slf4jLogConsumer(log, true).withPrefix("BACKEND-3")); - - - @EqualsAndHashCode - @ToString - class User - { - @Getter - private final String name; - private int serial = 0; - - - User (String name) - { - this.name = name; - } - - - int nextSerial() - { - return ++serial; - } - } - - @Getter - @Setter - static class StatusTo - { - String status; } }