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

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 e1e7813..aae931e 100644 (file)
--- a/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java
+++ b/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java
@@ -52,7 +52,7 @@ public abstract class AbstractHandoverIT
         .range(0, NUM_CLIENTS)
         .map(i -> new TestWriter(
             port,
-            chatRooms,
+            chatRooms[i % NUM_CHATROOMS],
             "user-" + Integer.toString(i)))
         .doOnNext(testClient -> executorService.execute(testClient))
         .toStream()
@@ -66,8 +66,9 @@ public abstract class AbstractHandoverIT
 
     Flux
         .fromArray(chatRooms)
-        .flatMap(chatRoom ->receiveMessages(chatRoom).take(50))
+        .flatMap(chatRoom ->receiveMessages(chatRoom))
         .doOnNext(message -> log.info("message: {}", message))
+        .take(50)
         .then()
         .block();
   }

diff --git a/src/test/java/de/juplo/kafka/chat/backend/TestWriter.java b/src/test/java/de/juplo/kafka/chat/backend/TestWriter.java
index 076376e..929d84f 100644 (file)
--- a/src/test/java/de/juplo/kafka/chat/backend/TestWriter.java
+++ b/src/test/java/de/juplo/kafka/chat/backend/TestWriter.java
@@ -24,7 +24,7 @@ public class TestWriter implements Runnable
     for (int i = 0; running; i++)
     {
       String message = "Message #" + i;
-      for (ChatRoomInfoTo chatRoom : chatRooms)
+      try
       {
         sendMessage(chatRoom, message)
             .retryWhen(Retry.fixedDelay(10, Duration.ofSeconds(1)))
@@ -39,9 +39,7 @@ public class TestWriter implements Runnable
                 user,
                 chatRoom,
                 result));
-      }
-      try
-      {
+
         Thread.sleep(ThreadLocalRandom.current().nextLong(700, 1000));
       }
       catch (Exception e)
@@ -80,16 +78,16 @@ public class TestWriter implements Runnable
 
 
   private final WebClient webClient;
-  private final ChatRoomInfoTo[] chatRooms;
+  private final ChatRoomInfoTo chatRoom;
   private final User user;
 
   volatile boolean running = true;
 
 
-  TestWriter(Integer port, ChatRoomInfoTo[] chatRooms, String username)
+  TestWriter(Integer port, ChatRoomInfoTo chatRoom, String username)
   {
     webClient = WebClient.create("http://localhost:" + port);
-    this.chatRooms = chatRooms;
+    this.chatRoom = chatRoom;
     user = new User(username);
   }
 }