Add test utils

Author: acogoluegnes

src/test/java/com/rabbitmq/stream/impl/StreamConsumerTest.java

@@ -13,12 +13,7 @@
 // [email protected].
 package com.rabbitmq.stream.impl;
 
-import static com.rabbitmq.stream.impl.TestUtils.b;
-import static com.rabbitmq.stream.impl.TestUtils.latchAssert;
-import static com.rabbitmq.stream.impl.TestUtils.localhost;
-import static com.rabbitmq.stream.impl.TestUtils.streamName;
-import static com.rabbitmq.stream.impl.TestUtils.waitAtMost;
-import static com.rabbitmq.stream.impl.TestUtils.waitMs;
+import static com.rabbitmq.stream.impl.TestUtils.*;
 import static java.lang.String.format;
 import static java.util.Collections.synchronizedList;
 import static org.assertj.core.api.Assertions.*;
@@ -227,23 +222,7 @@ void consumeWithAsyncConsumerFlowControl() throws Exception {
     Consumer consumer = consumerBuilder.build();
 
     waitAtMost(() -> receivedMessageCount.get() >= processingLimit);
-    int sameValueCount = 0;
-    Duration timeout = Duration.ofSeconds(10);
-    Duration waitTime = Duration.ofMillis(100);
-    long waitedTime = 0;
-    int lastValue = -1;
-    while (sameValueCount < 10) {
-      if (receivedMessageCount.get() == lastValue) {
-        sameValueCount++;
-      } else {
-        lastValue = receivedMessageCount.get();
-      }
-      Thread.sleep(waitTime.toMillis());
-      waitedTime += waitTime.toMillis();
-      if (waitedTime > timeout.toMillis()) {
-        fail("Consumption did not stop after %d seconds", timeout.getSeconds());
-      }
-    }
+    waitUntilStable(receivedMessageCount::get);
 
     assertThat(receivedMessageCount)
         .hasValueGreaterThanOrEqualTo(processingLimit)

src/test/java/com/rabbitmq/stream/impl/TestUtils.java

@@ -63,12 +63,10 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.Consumer;
-import java.util.function.Function;
-import java.util.function.Predicate;
-import java.util.function.Supplier;
+import java.util.function.*;
 import java.util.stream.IntStream;
 import org.assertj.core.api.AssertDelegateTarget;
+import org.assertj.core.api.Assertions;
 import org.assertj.core.api.Condition;
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.TestInfo;
@@ -966,4 +964,29 @@ static Client.ChunkListener credit() {
       return null;
     };
   }
+
+  static void waitUntilStable(LongSupplier value) {
+    int sameValueCount = 0;
+    Duration timeout = Duration.ofSeconds(10);
+    Duration waitTime = Duration.ofMillis(100);
+    long waitedTime = 0;
+    long lastValue = -1;
+    while (sameValueCount < 10) {
+      if (value.getAsLong() == lastValue) {
+        sameValueCount++;
+      } else {
+        lastValue = value.getAsLong();
+      }
+      try {
+        Thread.sleep(waitTime.toMillis());
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        throw new RuntimeException(e);
+      }
+      waitedTime += waitTime.toMillis();
+      if (waitedTime > timeout.toMillis()) {
+        Assertions.fail("Did not stabilize after %d seconds", timeout.getSeconds());
+      }
+    }
+  }
 }