Dispatch back to IO thread for non blocking requests

extensions/resteasy-reactive/quarkus-resteasy-reactive-kotlin/runtime/src/main/kotlin/org/jboss/resteasy/reactive/server/runtime/kotlin/FlowToPublisherHandler.kt

@@ -7,6 +7,7 @@ import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.launch
 import org.jboss.resteasy.reactive.server.core.ResteasyReactiveRequestContext
 import org.jboss.resteasy.reactive.server.spi.ServerRestHandler
+import java.util.concurrent.Executor
 import javax.enterprise.inject.spi.CDI
 
 class FlowToPublisherHandler : ServerRestHandler {
@@ -27,7 +28,9 @@ class FlowToPublisherHandler : ServerRestHandler {
                 // ensure the proper CL is not lost in dev-mode
                 Thread.currentThread().contextClassLoader = originalTCCL
                 requestContext.result = result.asMulti()
-                requestContext.resume()
+                //run in a direct invocation executor to run the rest of the invocation in the co-route scope
+                //feels a bit fragile, but let's see how it goes
+                requestContext.resume(Executor { it.run() })
             }
         }
     }

extensions/resteasy-reactive/quarkus-resteasy-reactive-servlet/runtime/src/main/java/io/quarkus/resteasy/reactive/server/servlet/runtime/ServletRequestContext.java

@@ -317,6 +317,12 @@ public <T> T unwrap(Class<T> theType) {
         return null;
     }
 
+    @Override
+    public boolean isOnIoThread() {
+        //does not really apply to Servlet
+        return true;
+    }
+
     @Override
     public ServerHttpResponse setStatusCode(int code) {
         response.setStatus(code);

independent-projects/resteasy-reactive/common/runtime/src/main/java/org/jboss/resteasy/reactive/common/core/AbstractResteasyReactiveContext.java

@@ -117,7 +117,7 @@ public Throwable getThrowable() {
 
     protected abstract Executor getEventLoop();
 
-    protected Executor getContextExecutor() {
+    public Executor getContextExecutor() {
         return null;
     }
 

independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/core/startup/RuntimeResourceDeployment.java

@@ -65,6 +65,7 @@
 import org.jboss.resteasy.reactive.server.handlers.InputHandler;
 import org.jboss.resteasy.reactive.server.handlers.InstanceHandler;
 import org.jboss.resteasy.reactive.server.handlers.InvocationHandler;
+import org.jboss.resteasy.reactive.server.handlers.NonBlockingHandler;
 import org.jboss.resteasy.reactive.server.handlers.ParameterHandler;
 import org.jboss.resteasy.reactive.server.handlers.PerRequestInstanceHandler;
 import org.jboss.resteasy.reactive.server.handlers.RequestDeserializeHandler;
@@ -184,6 +185,7 @@ public RuntimeResource buildResourceMethod(ResourceClass clazz,
                 blockingHandlerIndex = Optional.of(handlers.size() - 1);
                 score.add(ScoreSystem.Category.Execution, ScoreSystem.Diagnostic.ExecutionBlocking);
             } else {
+                handlers.add(NonBlockingHandler.INSTANCE);
                 score.add(ScoreSystem.Category.Execution, ScoreSystem.Diagnostic.ExecutionNonBlocking);
             }
         }

independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/handlers/NonBlockingHandler.java

@@ -0,0 +1,18 @@
+package org.jboss.resteasy.reactive.server.handlers;
+
+import org.jboss.resteasy.reactive.server.core.ResteasyReactiveRequestContext;
+import org.jboss.resteasy.reactive.server.spi.ServerRestHandler;
+
+public class NonBlockingHandler implements ServerRestHandler {
+
+    public static final NonBlockingHandler INSTANCE = new NonBlockingHandler();
+
+    @Override
+    public void handle(ResteasyReactiveRequestContext requestContext) throws Exception {
+        if (requestContext.serverRequest().isOnIoThread()) {
+            return;
+        }
+        requestContext.suspend();
+        requestContext.resume(requestContext.getContextExecutor());
+    }
+}

independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/spi/ServerHttpRequest.java

@@ -69,6 +69,8 @@ default FormData getExistingParsedForm() {
         return null;
     }
 
+    boolean isOnIoThread();
+
     interface ReadCallback {
 
         void done();

independent-projects/resteasy-reactive/server/vertx/src/main/java/org/jboss/resteasy/reactive/server/vertx/VertxResteasyReactiveRequestContext.java

@@ -66,7 +66,12 @@ public VertxResteasyReactiveRequestContext(Deployment deployment, ProvidersImpl
         this.contextExecutor = new Executor() {
             @Override
             public void execute(Runnable command) {
-                internal.execute(command);
+                internal.runOnContext(new Handler<Void>() {
+                    @Override
+                    public void handle(Void unused) {
+                        command.run();
+                    }
+                });
             }
         };
     }
@@ -101,7 +106,7 @@ protected EventLoop getEventLoop() {
         return ((ConnectionBase) context.request().connection()).channel().eventLoop();
     }
 
-    protected Executor getContextExecutor() {
+    public Executor getContextExecutor() {
         return contextExecutor;
     }
 
@@ -283,6 +288,11 @@ public FormData getExistingParsedForm() {
         return ret;
     }
 
+    @Override
+    public boolean isOnIoThread() {
+        return ((ConnectionBase) request.connection()).channel().eventLoop().inEventLoop();
+    }
+
     @SuppressWarnings("unchecked")
     @Override
     public <T> T unwrap(Class<T> theType) {

integration-tests/hibernate-reactive-postgresql/pom.xml

@@ -31,6 +31,10 @@
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-resteasy-reactive-jsonb</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-elytron-security-properties-file</artifactId>
+        </dependency>
 
         <!-- test dependencies -->
         <dependency>
@@ -84,6 +88,19 @@
                 </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-elytron-security-properties-file-deployment</artifactId>
+            <version>${project.version}</version>
+            <type>pom</type>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>*</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
     </dependencies>
 
     <build>

integration-tests/hibernate-reactive-postgresql/src/main/java/io/quarkus/it/hibernate/reactive/postgresql/HibernateReactiveTestEndpoint.java

@@ -6,13 +6,15 @@
 
 import org.hibernate.reactive.mutiny.Mutiny;
 
+import io.quarkus.security.Authenticated;
 import io.smallrye.mutiny.Uni;
 import io.vertx.mutiny.pgclient.PgPool;
 import io.vertx.mutiny.sqlclient.Row;
 import io.vertx.mutiny.sqlclient.RowSet;
 import io.vertx.mutiny.sqlclient.Tuple;
 
 @Path("/tests")
+@Authenticated
 public class HibernateReactiveTestEndpoint {
 
     @Inject

integration-tests/hibernate-reactive-postgresql/src/main/resources/application.properties

@@ -10,3 +10,8 @@ quarkus.hibernate-orm.database.generation=drop-and-create
 quarkus.datasource.reactive=true
 quarkus.datasource.reactive.url=${postgres.reactive.url}
 
+
+quarkus.security.users.embedded.enabled=true
+quarkus.security.users.embedded.users.scott=jb0ss
+quarkus.security.users.embedded.plain-text=true
+quarkus.security.users.embedded.roles.scott=Admin,admin,Tester,user

integration-tests/hibernate-reactive-postgresql/src/test/java/io/quarkus/it/hibernate/reactive/postgresql/HibernateReactiveTest.java

@@ -11,54 +11,62 @@
 
 /**
  * Test various JPA operations running in Quarkus
+ *
+ * Also makes sure that these work with a blocking security implementation
  */
 @QuarkusTest
 @TestHTTPEndpoint(HibernateReactiveTestEndpoint.class)
 public class HibernateReactiveTest {
 
     @Test
     public void reactiveCowPersist() {
-        RestAssured.when()
+        RestAssured.given().when()
+                .auth().preemptive().basic("scott", "jb0ss")
                 .get("/reactiveCowPersist")
                 .then()
                 .body(containsString("\"name\":\"Carolina\"}")); //Use containsString as we don't know the Id this object will have
     }
 
     @Test
     public void reactiveFindMutiny() {
-        RestAssured.when()
+        RestAssured.given().when()
+                .auth().preemptive().basic("scott", "jb0ss")
                 .get("/reactiveFindMutiny")
                 .then()
                 .body(is("{\"id\":5,\"name\":\"Aloi\"}"));
     }
 
     @Test
     public void reactivePersist() {
-        RestAssured.when()
+        RestAssured.given().when()
+                .auth().preemptive().basic("scott", "jb0ss")
                 .get("/reactivePersist")
                 .then()
                 .body(is("Tulip"));
     }
 
     @Test
     public void reactiveRemoveTransientEntity() {
-        RestAssured.when()
+        RestAssured.given().when()
+                .auth().preemptive().basic("scott", "jb0ss")
                 .get("/reactiveRemoveTransientEntity")
                 .then()
                 .body(is("OK"));
     }
 
     @Test
     public void reactiveRemoveManagedEntity() {
-        RestAssured.when()
+        RestAssured.given().when()
+                .auth().preemptive().basic("scott", "jb0ss")
                 .get("/reactiveRemoveManagedEntity")
                 .then()
                 .body(is("OK"));
     }
 
     @Test
     public void reactiveUpdate() {
-        RestAssured.when()
+        RestAssured.given().when()
+                .auth().preemptive().basic("scott", "jb0ss")
                 .get("/reactiveUpdate")
                 .then()
                 .body(is("Tina"));