Refactor Java tests after Client package change

3 files changed, 14 insertions, 10 deletions

diff --git a/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/FailureTest.java b/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/FailureTest.java
index 89f1073704..e5a0c9ba6d 100644
--- a/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/FailureTest.java
+++ b/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/FailureTest.java
@@ -17,6 +17,7 @@
 package com.rabbitmq.stream;
 
 import com.rabbitmq.stream.codec.WrapperMessageBuilder;
+import com.rabbitmq.stream.impl.Client;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -78,7 +79,8 @@ public class FailureTest {
                 .publishConfirmListener(publishingId -> confirmLatch.get().countDown()));
         String message = "all nodes available";
         messages.add(message);
-        publisher.publish(stream, message.getBytes(StandardCharsets.UTF_8));
+        publisher.publish(stream,
+                Collections.singletonList(publisher.messageBuilder().addData(message.getBytes(StandardCharsets.UTF_8)).build()));
         assertThat(confirmLatch.get().await(10, TimeUnit.SECONDS)).isTrue();
         confirmLatch.set(null);
 
@@ -102,7 +104,8 @@ public class FailureTest {
             confirmLatch.set(new CountDownLatch(1));
             message = "2 nodes available";
             messages.add(message);
-            publisher.publish(stream, message.getBytes(StandardCharsets.UTF_8));
+            publisher.publish(stream, Collections.singletonList(publisher.messageBuilder()
+                    .addData(message.getBytes(StandardCharsets.UTF_8)).build()));
             assertThat(confirmLatch.get().await(10, TimeUnit.SECONDS)).isTrue();
             confirmLatch.set(null);
         } finally {
@@ -118,7 +121,8 @@ public class FailureTest {
         confirmLatch.set(new CountDownLatch(1));
         message = "all nodes are back";
         messages.add(message);
-        publisher.publish(stream, message.getBytes(StandardCharsets.UTF_8));
+        publisher.publish(stream, Collections.singletonList(publisher.messageBuilder()
+                .addData(message.getBytes(StandardCharsets.UTF_8)).build()));
         assertThat(confirmLatch.get().await(10, TimeUnit.SECONDS)).isTrue();
         confirmLatch.set(null);
 
@@ -222,7 +226,7 @@ public class FailureTest {
                             .messageBuilder().applicationProperties().entry("generation", generation.get())
                             .messageBuilder().build();
                     try {
-                        long publishingId = publisher.get().publish(stream, message);
+                        long publishingId = publisher.get().publish(stream, Collections.singletonList(message)).get(0);
                         published.put(publishingId, message);
                     } catch (Exception e) {
                         // keep going
@@ -345,7 +349,7 @@ public class FailureTest {
                         .messageBuilder().applicationProperties().entry("generation", generation.get())
                         .messageBuilder().build();
                 try {
-                    long publishingId = publisher.publish(stream, message);
+                    long publishingId = publisher.publish(stream, Collections.singletonList(message)).get(0);
                     published.put(publishingId, message);
                 } catch (Exception e) {
                     // keep going
diff --git a/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/StreamTest.java b/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/StreamTest.java
index d3aaec6567..b15c27119a 100644
--- a/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/StreamTest.java
+++ b/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/StreamTest.java
@@ -16,6 +16,7 @@
 
 package com.rabbitmq.stream;
 
+import com.rabbitmq.stream.impl.Client;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -23,10 +24,7 @@ import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
 
 import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -73,7 +71,8 @@ public class StreamTest {
                 .port(publisherBroker.apply(streamMetadata).getPort())
                 .publishConfirmListener(publishingId -> publishingLatch.countDown()));
 
-        IntStream.range(0, messageCount).forEach(i -> publisher.publish(stream, ("hello " + i).getBytes(StandardCharsets.UTF_8)));
+        IntStream.range(0, messageCount).forEach(i -> publisher.publish(stream, Collections.singletonList(
+                publisher.messageBuilder().addData(("hello " + i).getBytes(StandardCharsets.UTF_8)).build())));
 
         assertThat(publishingLatch.await(10, TimeUnit.SECONDS)).isTrue();
 
diff --git a/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/TestUtils.java b/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/TestUtils.java
index 5f625ef1c5..da3eb28f02 100644
--- a/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/TestUtils.java
+++ b/deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/TestUtils.java
@@ -16,6 +16,7 @@
 
 package com.rabbitmq.stream;
 
+import com.rabbitmq.stream.impl.Client;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import org.junit.jupiter.api.extension.*;