Fix #33106.

In addition to Multi, also handle the Flow.Publisher case.
quarkusio · May 5, 2023 · 3cbe472 · 3cbe472
1 parent 71a14c4
commit 3cbe472
Show file tree

Hide file tree

Showing 3 changed files with 43 additions and 2 deletions.
diff --git a/...c/main/java/io/quarkus/resteasy/reactive/server/deployment/ResteasyReactiveProcessor.java b/...c/main/java/io/quarkus/resteasy/reactive/server/deployment/ResteasyReactiveProcessor.java
@@ -4,8 +4,7 @@
 import static io.quarkus.resteasy.reactive.common.deployment.QuarkusResteasyReactiveDotNames.HTTP_SERVER_RESPONSE;
 import static io.quarkus.resteasy.reactive.common.deployment.QuarkusResteasyReactiveDotNames.ROUTING_CONTEXT;
 import static java.util.stream.Collectors.toList;
-import static org.jboss.resteasy.reactive.common.processor.ResteasyReactiveDotNames.DATE_FORMAT;
-import static org.jboss.resteasy.reactive.common.processor.ResteasyReactiveDotNames.MULTI;
+import static org.jboss.resteasy.reactive.common.processor.ResteasyReactiveDotNames.*;
 
 import java.io.File;
 import java.io.IOException;
@@ -552,6 +551,7 @@ public void accept(EndpointIndexer.ResourceMethodCallbackEntry entry) {
 
                             if (paramsRequireReflection ||
                                     MULTI.toString().equals(entry.getResourceMethod().getSimpleReturnType()) ||
+                                    PUBLISHER.toString().equals(entry.getResourceMethod().getSimpleReturnType()) ||
                                     filtersAccessResourceMethod(resourceInterceptorsBuildItem.getResourceInterceptors()) ||
                                     entry.additionalRegisterClassForReflectionCheck()) {
                                 minimallyRegisterResourceClassForReflection(entry, reflectiveClassBuildItemBuildProducer);

diff --git a/...ion-tests/resteasy-mutiny/src/main/java/io/quarkus/it/resteasy/mutiny/MutinyResource.java b/...ion-tests/resteasy-mutiny/src/main/java/io/quarkus/it/resteasy/mutiny/MutinyResource.java
@@ -1,6 +1,7 @@
 package io.quarkus.it.resteasy.mutiny;
 
 import java.io.IOException;
+import java.util.concurrent.Flow;
 
 import jakarta.inject.Inject;
 import jakarta.ws.rs.GET;
@@ -66,6 +67,14 @@ public Multi<Pet> sse() {
         return service.getMorePets();
     }
 
+    @GET
+    @Path("/pets/flow")
+    @Produces(MediaType.SERVER_SENT_EVENTS)
+    @RestStreamElementType(MediaType.APPLICATION_JSON)
+    public Flow.Publisher<Pet> sseFlow() {
+        return service.getMorePets();
+    }
+
     @Inject
     @RestClient
     MyRestService client;

diff --git a/...gration-tests/resteasy-mutiny/src/test/java/io/quarkus/it/resteasy/mutiny/MutinyTest.java b/...gration-tests/resteasy-mutiny/src/test/java/io/quarkus/it/resteasy/mutiny/MutinyTest.java
@@ -111,6 +111,38 @@ public void accept(UniEmitter<? super List<Pet>> uniEmitter) {
         }
     }
 
+    @Test
+    public void testSSEWithFlowPublisher() {
+        Client client = ClientBuilder.newClient();
+        WebTarget target = client.target("http://localhost:" + RestAssured.port + "/mutiny/pets/flow");
+        try (SseEventSource eventSource = SseEventSource.target(target).build()) {
+            Uni<List<Pet>> petList = Uni.createFrom().emitter(new Consumer<UniEmitter<? super List<Pet>>>() {
+                @Override
+                public void accept(UniEmitter<? super List<Pet>> uniEmitter) {
+                    List<Pet> pets = new CopyOnWriteArrayList<>();
+                    eventSource.register(event -> {
+                        Pet pet = event.readData(Pet.class, MediaType.APPLICATION_JSON_TYPE);
+                        pets.add(pet);
+                        if (pets.size() == 5) {
+                            uniEmitter.complete(pets);
+                        }
+                    }, ex -> {
+                        uniEmitter.fail(new IllegalStateException("SSE failure", ex));
+                    });
+                    eventSource.open();
+
+                }
+            });
+            List<Pet> pets = petList.await().atMost(Duration.ofMinutes(1));
+            Assertions.assertEquals(5, pets.size());
+            Assertions.assertEquals("neo", pets.get(0).getName());
+            Assertions.assertEquals("indy", pets.get(1).getName());
+            Assertions.assertEquals("plume", pets.get(2).getName());
+            Assertions.assertEquals("titi", pets.get(3).getName());
+            Assertions.assertEquals("rex", pets.get(4).getName());
+        }
+    }
+
     @Test
     public void testClientReturningUni() {
         get("/mutiny/client")