RESTEasy Reactive thread fixes

extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/BasicRestClientTest.java

@@ -45,4 +45,9 @@ void shouldHaveApplicationScopeByDefault() {
         Bean<?> resolvedBean = beanManager.resolve(beans);
         assertThat(resolvedBean.getScope()).isEqualTo(ApplicationScoped.class);
     }
+
+    @Test
+    void shouldInvokeClientResponseOnSameContext() {
+        assertThat(testBean.bug18977()).isEqualTo("Hello");
+    }
 }

extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/HelloClient2.java

@@ -1,16 +1,26 @@
 package io.quarkus.rest.client.reactive;
 
 import javax.ws.rs.Consumes;
+import javax.ws.rs.GET;
 import javax.ws.rs.POST;
 import javax.ws.rs.Path;
 import javax.ws.rs.core.MediaType;
 
 import org.eclipse.microprofile.rest.client.inject.RegisterRestClient;
 
+import io.smallrye.mutiny.Uni;
+
 @RegisterRestClient(configKey = "hello2")
 public interface HelloClient2 {
     @POST
     @Consumes(MediaType.TEXT_PLAIN)
     @Path("/")
     String echo(String name);
+
+    @GET
+    String bug18977();
+
+    @GET
+    @Path("delay")
+    Uni<String> delay();
 }

extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/HelloResource.java

@@ -1,13 +1,21 @@
 package io.quarkus.rest.client.reactive;
 
+import java.time.Duration;
+
 import javax.ws.rs.Consumes;
+import javax.ws.rs.GET;
 import javax.ws.rs.POST;
 import javax.ws.rs.Path;
 import javax.ws.rs.Produces;
 import javax.ws.rs.core.Context;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Request;
 
+import org.eclipse.microprofile.rest.client.inject.RestClient;
+import org.junit.jupiter.api.Assertions;
+
+import io.smallrye.mutiny.Uni;
+
 @Path("/hello")
 @Produces(MediaType.TEXT_PLAIN)
 @Consumes(MediaType.TEXT_PLAIN)
@@ -16,4 +24,24 @@ public class HelloResource {
     public String echo(String name, @Context Request request) {
         return "hello, " + name;
     }
+
+    @RestClient
+    HelloClient2 client2;
+
+    @GET
+    public Uni<String> something() {
+        Thread thread = Thread.currentThread();
+        return client2.delay()
+                .map(foo -> {
+                    Assertions.assertSame(thread, Thread.currentThread());
+                    return foo;
+                });
+    }
+
+    @Path("delay")
+    @GET
+    public Uni<String> delay() {
+        return Uni.createFrom().item("Hello")
+                .onItem().delayIt().by(Duration.ofMillis(500));
+    }
 }

extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/TestBean.java

@@ -27,4 +27,8 @@ String helloViaBuiltClient(String name) {
                 .build(HelloClient.class);
         return helloClient.echo(name);
     }
+
+    String bug18977() {
+        return client2.bug18977();
+    }
 }

extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/provider/ProviderDisabledAutodiscoveryTest.java

@@ -15,6 +15,7 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 
+import io.quarkus.rest.client.reactive.HelloClient2;
 import io.quarkus.rest.client.reactive.HelloResource;
 import io.quarkus.test.QuarkusUnitTest;
 import io.quarkus.test.common.http.TestHTTPResource;
@@ -24,9 +25,10 @@ public class ProviderDisabledAutodiscoveryTest {
     @RegisterExtension
     static final QuarkusUnitTest config = new QuarkusUnitTest()
             .setArchiveProducer(() -> ShrinkWrap.create(JavaArchive.class)
-                    .addClasses(HelloResource.class, HelloClient.class, GlobalRequestFilter.class, GlobalResponseFilter.class)
+                    .addClasses(HelloResource.class, HelloClient.class, HelloClient2.class, GlobalRequestFilter.class,
+                            GlobalResponseFilter.class)
                     .addAsResource(
-                            new StringAsset(setUrlForClass(HelloClient.class)
+                            new StringAsset(setUrlForClass(HelloClient.class) + setUrlForClass(HelloClient2.class)
                                     + "quarkus.rest-client-reactive.provider-autodiscovery=false"),
                             "application.properties"));
 

extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/provider/ProviderPriorityTest.java

@@ -13,6 +13,7 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 
+import io.quarkus.rest.client.reactive.HelloClient2;
 import io.quarkus.rest.client.reactive.HelloResource;
 import io.quarkus.test.QuarkusUnitTest;
 
@@ -23,12 +24,13 @@ public class ProviderPriorityTest {
             .setArchiveProducer(() -> ShrinkWrap.create(JavaArchive.class)
                     .addClasses(HelloResource.class,
                             HelloClient.class,
+                            HelloClient2.class,
                             HelloClientWithFilter.class,
                             ResponseFilterLowestPrio.class,
                             GlobalResponseFilter.class,
                             GlobalResponseFilterLowPrio.class)
                     .addAsResource(
-                            new StringAsset(setUrlForClass(HelloClient.class)
+                            new StringAsset(setUrlForClass(HelloClient.class) + setUrlForClass(HelloClient2.class)
                                     + setUrlForClass(HelloClientWithFilter.class)),
                             "application.properties"));
 

extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/provider/ProviderTest.java

@@ -16,6 +16,7 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 
+import io.quarkus.rest.client.reactive.HelloClient2;
 import io.quarkus.rest.client.reactive.HelloResource;
 import io.quarkus.test.QuarkusUnitTest;
 import io.quarkus.test.common.http.TestHTTPResource;
@@ -25,11 +26,11 @@ public class ProviderTest {
     @RegisterExtension
     static final QuarkusUnitTest config = new QuarkusUnitTest()
             .setArchiveProducer(() -> ShrinkWrap.create(JavaArchive.class)
-                    .addClasses(HelloResource.class, HelloClient.class, GlobalRequestFilter.class,
+                    .addClasses(HelloResource.class, HelloClient.class, GlobalRequestFilter.class, HelloClient2.class,
                             GlobalResponseFilter.class, GlobalRequestFilterConstrainedToServer.class,
                             GlobalFeature.class)
                     .addAsResource(
-                            new StringAsset(setUrlForClass(HelloClient.class)),
+                            new StringAsset(setUrlForClass(HelloClient.class) + setUrlForClass(HelloClient2.class)),
                             "application.properties"));
 
     @RestClient

independent-projects/resteasy-reactive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/handlers/ClientSendRequestHandler.java

@@ -59,6 +59,7 @@ public void handle(Throwable event) {
         future.onSuccess(new Handler<HttpClientRequest>() {
             @Override
             public void handle(HttpClientRequest httpClientRequest) {
+                requestContext.setHttpClientRequest(httpClientRequest);
                 Future<HttpClientResponse> sent;
                 if (requestContext.isMultipart()) {
                     Promise<HttpClientRequest> requestPromise = Promise.promise();

independent-projects/resteasy-reactive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/RestClientRequestContext.java

@@ -1,10 +1,13 @@
 package org.jboss.resteasy.reactive.client.impl;
 
+import io.vertx.core.Context;
 import io.vertx.core.MultiMap;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.http.HttpClient;
 import io.vertx.core.http.HttpClientRequest;
 import io.vertx.core.http.HttpClientResponse;
+import io.vertx.core.impl.ContextInternal;
+import io.vertx.core.net.impl.ConnectionBase;
 import io.vertx.ext.web.multipart.MultipartForm;
 import java.io.IOException;
 import java.io.InputStream;
@@ -153,6 +156,17 @@ public ClientResponseContextImpl getOrCreateClientResponseContext() {
         return clientResponseContext;
     }
 
+    @Override
+    protected Executor getContextExecutor() {
+        if (Context.isOnEventLoopThread()) {
+            return ((ContextInternal) restClient.getVertx().getOrCreateContext()).nettyEventLoop();
+        }
+        if (httpClientRequest != null) {
+            return ((ConnectionBase) httpClientRequest.connection()).getContext().nettyEventLoop();
+        }
+        return null;
+    }
+
     public ClientRequestContextImpl getOrCreateClientRequestContext() {
         if (clientRequestContext == null) {
             clientRequestContext = new ClientRequestContextImpl(this, this.getRestClient(), this.getConfiguration());
@@ -201,15 +215,20 @@ public HttpClientResponse getVertxClientResponse() {
 
     @Override
     protected Executor getEventLoop() {
-        if (httpClientRequest == null) {
-            return restClient.getVertx().nettyEventLoopGroup().next();
-        } else {
+        if (Context.isOnEventLoopThread()) {
+            Context context = restClient.getVertx().getOrCreateContext();
             return new Executor() {
                 @Override
                 public void execute(Runnable command) {
-                    command.run();
+                    context.runOnContext(v -> command.run());
                 }
             };
+        } else if (httpClientRequest == null) {
+            //we are on a worker and have no bound event loop
+            //this should never really happen
+            return restClient.getVertx().nettyEventLoopGroup().next();
+        } else {
+            return ((ConnectionBase) httpClientRequest.connection()).channel().eventLoop();
         }
     }