RR client: run client on same context

Fixes quarkusio#18977

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();
+    }
 }

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

@@ -1,5 +1,6 @@
 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;
@@ -219,7 +220,14 @@ public HttpClientResponse getVertxClientResponse() {
     @Override
     protected Executor getEventLoop() {
         if (httpClientRequest == null) {
-            return restClient.getVertx().nettyEventLoopGroup().next();
+            // make sure we execute the client callbacks on the same context as the current thread
+            Context context = restClient.getVertx().getOrCreateContext();
+            return new Executor() {
+                @Override
+                public void execute(Runnable command) {
+                    context.runOnContext(v -> command.run());
+                }
+            };
         } else {
             return new Executor() {
                 @Override