Making quarkus kafka test use reactive jaxrs client

Author: edeandrea

chapter-5/chapter-5-quarkus-kafka-streams/pom.xml

@@ -47,23 +47,13 @@
       <scope>test</scope>
     </dependency>
     <dependency>
-      <groupId>org.jboss.resteasy</groupId>
-      <artifactId>resteasy-client</artifactId>
-      <scope>test</scope>
-    </dependency>
-    <dependency>
-      <groupId>io.rest-assured</groupId>
-      <artifactId>rest-assured</artifactId>
-      <scope>test</scope>
-    </dependency>
-    <dependency>
-      <groupId>org.awaitility</groupId>
-      <artifactId>awaitility</artifactId>
+      <groupId>org.assertj</groupId>
+      <artifactId>assertj-core</artifactId>
       <scope>test</scope>
     </dependency>
     <dependency>
-      <groupId>org.assertj</groupId>
-      <artifactId>assertj-core</artifactId>
+      <groupId>io.quarkus</groupId>
+      <artifactId>quarkus-jaxrs-client-reactive</artifactId>
       <scope>test</scope>
     </dependency>
   </dependencies>

chapter-5/chapter-5-quarkus-kafka-streams/src/test/java/org/acme/rest/PriceResourceTest.java

@@ -1,23 +1,21 @@
 package org.acme.rest;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.awaitility.Awaitility.await;
 
 import java.net.URI;
 import java.time.Duration;
 import java.util.List;
-import java.util.concurrent.CopyOnWriteArrayList;
 
-import javax.ws.rs.client.Client;
 import javax.ws.rs.client.ClientBuilder;
-import javax.ws.rs.client.WebTarget;
-import javax.ws.rs.sse.SseEventSource;
+import javax.ws.rs.core.MediaType;
 
+import org.jboss.resteasy.reactive.client.impl.MultiInvoker;
 import org.junit.jupiter.api.Test;
 
 import io.quarkus.test.common.http.TestHTTPEndpoint;
 import io.quarkus.test.common.http.TestHTTPResource;
 import io.quarkus.test.junit.QuarkusTest;
+import io.smallrye.mutiny.helpers.test.AssertSubscriber;
 
 @QuarkusTest
 class PriceResourceTest {
@@ -26,20 +24,18 @@ class PriceResourceTest {
     URI uri;
 
     @Test
-    void testPricesEventStream() {
-        Client client = ClientBuilder.newClient();
-        WebTarget target = client.target(this.uri);
-
-        List<Double> received = new CopyOnWriteArrayList<>();
-
-        try (SseEventSource source = SseEventSource.target(target).build()) {
-            source.register(inboundSseEvent -> received.add(Double.valueOf(inboundSseEvent.readData())));
-            source.open();
-
-            await()
-              .atMost(Duration.ofSeconds(20))
-              .until(() -> received.size() == 3);
-        }
+    public void sseEventStream() {
+        List<Double> received = ClientBuilder.newClient()
+          .target(this.uri)
+          .request(MediaType.SERVER_SENT_EVENTS)
+          .rx(MultiInvoker.class)
+          .get(Double.class)
+          .select().first(3)
+          .subscribe().withSubscriber(AssertSubscriber.create(3))
+          .assertSubscribed()
+          .awaitItems(3, Duration.ofSeconds(20))
+          .assertCompleted()
+          .getItems();
 
         assertThat(received)
           .hasSize(3)