From: Kai Moritz <kai@juplo.de>
Date: Wed, 28 Feb 2024 08:59:20 +0000 (+0100)
Subject: test: HandoverIT-POC - running multiple clients in parallel
X-Git-Tag: rebase--2024-03-03--11-38~24
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=89691f0d15b35f8603ea649f07a727ab20ea73e1;p=demos%2Fkafka%2Fchat

test: HandoverIT-POC - running multiple clients in parallel
---

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 900a7c39..a9148f0d 100644
--- a/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java
+++ b/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java
@@ -13,6 +13,10 @@ import org.testcontainers.junit.jupiter.Testcontainers;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.stream.IntStream;
+
 
 @Testcontainers
 @Slf4j
@@ -20,9 +24,11 @@ public abstract class AbstractHandoverIT
 {
   static final ParameterizedTypeReference<ServerSentEvent<String>> 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)
@@ -40,11 +46,17 @@ public abstract class AbstractHandoverIT
         .toStream()
         .toArray(size -> new ChatRoomInfoTo[size]);
 
-    TestClient testClient = new TestClient(
-        containers.haproxy.getMappedPort(8400),
-        chatRooms,
-        "nerd");
-    testClient.run();
+    TestClient[] testClients = Flux
+        .fromStream(IntStream.range(0, NUM_CLIENTS).mapToObj(i -> 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(10000);
 
     Flux
         .fromArray(chatRooms)
diff --git a/src/test/java/de/juplo/kafka/chat/backend/TestClient.java b/src/test/java/de/juplo/kafka/chat/backend/TestClient.java
index 0461970f..964a06ba 100644
--- a/src/test/java/de/juplo/kafka/chat/backend/TestClient.java
+++ b/src/test/java/de/juplo/kafka/chat/backend/TestClient.java
@@ -14,8 +14,9 @@ import java.util.concurrent.ThreadLocalRandom;
 
 
 @Slf4j
-public class TestClient
+public class TestClient implements Runnable
 {
+  @Override
   public void run()
   {
     for (int i = 0; i < 100; i++)