test: HandoverIT-POC - Each `TestWriter` writes only to one chat-room

[demos/kafka/chat] / src / test / java / de / juplo / kafka / chat / backend / AbstractHandoverIT.java

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 a9148f0..41d9e49 100644 (file)
--- a/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java
+++ b/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java
@@ -13,9 +13,9 @@ import org.testcontainers.junit.jupiter.Testcontainers;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import java.util.Arrays;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.stream.IntStream;
 
 
 @Testcontainers
@@ -42,21 +42,26 @@ public abstract class AbstractHandoverIT
   {
     ChatRoomInfoTo[] chatRooms = Flux
         .range(0, NUM_CHATROOMS)
-        .flatMap(i -> createChatRoom("#" + i))
+        .flatMap(i -> createChatRoom("room-" + i))
         .toStream()
         .toArray(size -> new ChatRoomInfoTo[size]);
 
-    TestClient[] testClients = Flux
-        .fromStream(IntStream.range(0, NUM_CLIENTS).mapToObj(i -> Integer.toString(i)))
-        .map(i -> new TestClient(
-            containers.haproxy.getMappedPort(8400),
-            chatRooms,
-            i))
+    int port = containers.haproxy.getMappedPort(8400);
+
+    TestWriter[] testWriters = Flux
+        .range(0, NUM_CLIENTS)
+        .map(i -> new TestWriter(
+            port,
+            chatRooms[i % NUM_CHATROOMS],
+            "user-" + Integer.toString(i)))
         .doOnNext(testClient -> executorService.execute(testClient))
         .toStream()
-        .toArray(size -> new TestClient[size]);
+        .toArray(size -> new TestWriter[size]);
 
     Thread.sleep(10000);
+    Arrays
+        .stream(testWriters)
+        .forEach(testClient -> testClient.running = false);
 
     Flux
         .fromArray(chatRooms)