From: Kai Moritz <kai@juplo.de>
Date: Sun, 3 Mar 2024 09:20:11 +0000 (+0100)
Subject: WIP:test: HandoverIT-POC - Excuting Flux...
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=9b562633f242b662c7041a84e2a8057ecb3ac4cf;p=demos%2Fkafka%2Fchat

WIP:test: HandoverIT-POC - Excuting Flux...

* Droped the waiting for `TestListener` alltogehter.
* The waiting can be droped, because waiting for the `TestWriter`-instances
  ensures, that all messages are send (and therefore very likely received)
---

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 53085f77..c81b271f 100644
--- a/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java
+++ b/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java
@@ -57,9 +57,11 @@ public abstract class AbstractHandoverIT
     }
 
     TestListener testListener = new TestListener(port, chatRooms);
-    CompletableFuture<Void> testListenerFuture = testListener
+    testListener
         .run()
-        .toFuture();
+        .subscribe(message -> log.info(
+            "Received message: {}",
+            message));
 
     log.info("Sleeping for 2 seconds...");
     Thread.sleep(2000);
@@ -71,13 +73,8 @@ public abstract class AbstractHandoverIT
       log.info("Joined TestWriter {}", testWriters[i].user);
     }
 
-
-    log.info("Sleeping for 2 seconds...");
-    Thread.sleep(2000);
-    log.info("Joining TestListener...");
-    testListener.running = false;
-    testListenerFuture.join();
-    log.info("Joined TestListener");
+    // Yield the work, so that the last messages can be received
+    Thread.sleep(500);
   }
 
   Mono<ChatRoomInfoTo> createChatRoom(String name)
diff --git a/src/test/java/de/juplo/kafka/chat/backend/TestListener.java b/src/test/java/de/juplo/kafka/chat/backend/TestListener.java
index 35f65acd..78d4c82f 100644
--- a/src/test/java/de/juplo/kafka/chat/backend/TestListener.java
+++ b/src/test/java/de/juplo/kafka/chat/backend/TestListener.java
@@ -12,7 +12,6 @@ import org.springframework.http.codec.ServerSentEvent;
 import org.springframework.web.reactive.function.client.WebClient;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Schedulers;
 
 import java.util.*;
 
@@ -23,7 +22,7 @@ public class TestListener
   static final ParameterizedTypeReference<ServerSentEvent<String>> SSE_TYPE = new ParameterizedTypeReference<>() {};
 
 
-  public Mono<Void> run()
+  public Flux<MessageTo> run()
   {
     return Flux
         .fromArray(chatRooms)
@@ -44,21 +43,8 @@ public class TestListener
                   return Mono.error(e);
                 }
               })
-              .doOnNext(message ->
-              {
-                list.add(message);
-                log.info(
-                    "Received a message from chat-room {}: {}",
-                    chatRoom.getName(),
-                    message);
-              });
-        })
-        .limitRate(10)
-        .takeUntil(message -> !running)
-        .doOnComplete(() -> log.info("TestListener is done"))
-        .parallel(chatRooms.length)
-        .runOn(Schedulers.parallel())
-        .then();
+              .doOnNext(message -> list.add(message));
+        });
   }
 
   Flux<ServerSentEvent<String>> receiveMessages(ChatRoomInfoTo chatRoom)
@@ -80,8 +66,6 @@ public class TestListener
 
   final Map<UUID, List<MessageTo>> receivedMessages = new HashMap<>();
 
-  volatile boolean running = true;
-
 
   TestListener(Integer port, ChatRoomInfoTo[] chatRooms)
   {
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 0d2c41e7..fd339411 100644
--- a/src/test/java/de/juplo/kafka/chat/backend/TestWriter.java
+++ b/src/test/java/de/juplo/kafka/chat/backend/TestWriter.java
@@ -9,7 +9,6 @@ import org.springframework.web.reactive.function.client.WebClient;
 import org.springframework.web.reactive.function.client.WebClientResponseException;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Schedulers;
 import reactor.util.retry.Retry;
 
 import java.nio.charset.Charset;
@@ -63,11 +62,8 @@ public class TestWriter
               user,
               e.getResponseBodyAsString(Charset.defaultCharset()));
         })
-        .limitRate(1)
         .takeUntil(message -> !running)
         .doOnComplete(() -> log.info("TestWriter {} is done", user))
-        .parallel(1)
-        .runOn(Schedulers.parallel())
         .then();
   }