package de.juplo.kafka.chat.backend;
-import de.juplo.kafka.chat.backend.api.ChatRoomInfoTo;
-import de.juplo.kafka.chat.backend.api.MessageTo;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
import org.testcontainers.junit.jupiter.Testcontainers;
-import reactor.util.retry.Retry;
-
-import java.time.Duration;
-import java.util.stream.IntStream;
@Testcontainers
private final AbstractContainerTemplates containerTemplates;
- @Test
- void test() throws InterruptedException
+ AbstractHandoverIT(AbstractContainerTemplates containerTemplates)
{
- 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)
- .doOnNext(message -> log.info("message: {}", message))
- .then()
- .block();
+ this.containerTemplates = containerTemplates;
}
+
@BeforeEach
void setUp() throws Exception
{
package de.juplo.kafka.chat.backend;
import lombok.extern.slf4j.Slf4j;
-import org.testcontainers.containers.*;
-import org.testcontainers.containers.output.Slf4jLogConsumer;
-import org.testcontainers.containers.wait.strategy.Wait;
-import org.testcontainers.utility.DockerImageName;
-
-import java.io.IOException;
@Slf4j
class KafkaHandoverIT extends AbstractHandoverIT
{
- @Override
- void setUpExtra() throws IOException, InterruptedException
- {
- kafka.start();
-
- Container.ExecResult result;
- result = kafka.execInContainer(
- "kafka-topics",
- "--bootstrap-server",
- "kafka:9999",
- "--create",
- "--topic",
- "info_channel",
- "--partitions",
- "3");
- log.info(
- "EXIT-CODE={}, STDOUT={}, STDERR={}",
- result.getExitCode(),
- result.getStdout(),
- result.getStdout());
- result = kafka.execInContainer(
- "kafka-topics",
- "--bootstrap-server",
- "kafka:9999",
- "--create",
- "--topic",
- "data_channel",
- "--partitions",
- "10");
- log.info(
- "EXIT-CODE={}, STDOUT={}, STDERR={}",
- result.getExitCode(),
- result.getStdout(),
- result.getStdout());
- }
-
-
- KafkaContainer kafka =
- new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.4.0"))
- .withNetwork(network)
- .withNetworkAliases("kafka")
- .withListener(() -> "kafka:9999")
- .withKraft()
- .waitingFor(Wait.forLogMessage(".*Kafka\\ Server\\ started.*\\n", 1))
- .withLogConsumer(new Slf4jLogConsumer(log, true).withPrefix("KAFKA"));
-
- @Override
- String[] getCommandBackend1()
- {
- return new String[]
- {
- "--chat.backend.instance-id=backend-1",
- "--chat.backend.services=kafka",
- "--chat.backend.kafka.bootstrap-servers=kafka:9999",
- "--chat.backend.kafka.instance-uri=http://backend-1:8080",
- "--chat.backend.kafka.num-partitions=10",
- "--chat.backend.kafka.client-id-prefix=B1",
- "--chat.backend.kafka.haproxy-runtime-api=haproxy:8401",
- "--chat.backend.kafka.haproxy-map=/usr/local/etc/haproxy/sharding.map"
- };
- }
-
- @Override
- String[] getCommandBackend2()
- {
- return new String[]
- {
- "--chat.backend.instance-id=backend-2",
- "--chat.backend.services=kafka",
- "--chat.backend.kafka.bootstrap-servers=kafka:9999",
- "--chat.backend.kafka.instance-uri=http://backend-2:8080",
- "--chat.backend.kafka.num-partitions=10",
- "--chat.backend.kafka.client-id-prefix=B2",
- "--chat.backend.kafka.haproxy-runtime-api=haproxy:8401",
- "--chat.backend.kafka.haproxy-map=/usr/local/etc/haproxy/sharding.map"
- };
- }
-
- @Override
- String[] getCommandBackend3()
+ KafkaHandoverIT()
{
- return new String[]
- {
- "--chat.backend.instance-id=backend-3",
- "--chat.backend.services=kafka",
- "--chat.backend.kafka.bootstrap-servers=kafka:9999",
- "--chat.backend.kafka.instance-uri=http://backend-3:8080",
- "--chat.backend.kafka.num-partitions=10",
- "--chat.backend.kafka.client-id-prefix=B3",
- "--chat.backend.kafka.haproxy-runtime-api=haproxy:8401",
- "--chat.backend.kafka.haproxy-map=/usr/local/etc/haproxy/sharding.map"
- };
+ super(new KafkaContainerTemplates());
}
}