Support cancelling pending REST requests

Closes: #48620

Author: geoand

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

@@ -0,0 +1,95 @@
+package io.quarkus.rest.client.reactive;
+
+import static io.restassured.RestAssured.when;
+import static org.hamcrest.Matchers.is;
+
+import java.net.URI;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.IntStream;
+
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.Path;
+
+import org.eclipse.microprofile.rest.client.RestClientBuilder;
+import org.jboss.resteasy.reactive.client.api.QuarkusRestClientProperties;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.test.QuarkusUnitTest;
+import io.quarkus.test.common.http.TestHTTPResource;
+import io.smallrye.mutiny.Uni;
+
+public class RequestCancellationTest {
+
+    @RegisterExtension
+    static final QuarkusUnitTest config = new QuarkusUnitTest()
+            .withApplicationRoot((jar) -> jar.addClasses(Client.class, Resource.class));
+
+    @TestHTTPResource
+    URI uri;
+
+    @Test
+    public void test() throws InterruptedException {
+        Client client = RestClientBuilder.newBuilder()
+                .baseUri(uri)
+                .property(QuarkusRestClientProperties.CONNECTION_POOL_SIZE, 1) // make sure client requests are serialized
+                .build(Client.class);
+
+        when().get("resource/count")
+                .then()
+                .statusCode(200)
+                .body(is("0"));
+
+        CountDownLatch latch = new CountDownLatch(1);
+        client.get().subscribe().with(res -> {
+            latch.countDown();
+        });
+
+        // create a bunch of requests that we test won't end up hitting the server
+        Uni.join().all(IntStream.range(0, 100).mapToObj(i -> client.get()).toList())
+                .andCollectFailures()
+                .subscribe() // actually initiate the requests
+                .with(res -> {
+                })
+                .cancel(); // cancel all the requests
+
+        latch.await(5, TimeUnit.SECONDS);
+
+        // ensure that only the first request was made
+        when().get("resource/count")
+                .then()
+                .statusCode(200)
+                .body(is("1"));
+    }
+
+    @Path("resource")
+    public interface Client {
+
+        @GET
+        Uni<String> get();
+    }
+
+    @Path("resource")
+    public static class Resource {
+
+        private static final AtomicLong COUNTER = new AtomicLong();
+
+        @GET
+        public String get() throws InterruptedException {
+            COUNTER.incrementAndGet();
+            // ensure that each request takes a long time to complete to we don't end up with a race
+            // condition where the client requests that were to be canceled, had time to execute because
+            // the previous requests in the queue completed too fast
+            Thread.sleep(2000);
+            return "foo";
+        }
+
+        @Path("count")
+        @GET
+        public long count() {
+            return COUNTER.get();
+        }
+    }
+}

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

@@ -101,6 +101,11 @@ public void handle(RestClientRequestContext requestContext) {
         future.subscribe().with(new Consumer<>() {
             @Override
             public void accept(HttpClientRequest httpClientRequest) {
+                if (requestContext.isUserCanceled()) {
+                    // in this case the user aborted before the request was even created
+                    return;
+                }
+
                 requestContext.setHttpClientRequest(httpClientRequest);
 
                 // adapt headers to HTTP/2 depending on the underlying HTTP connection

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

@@ -15,6 +15,7 @@
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import jakarta.ws.rs.RuntimeType;
 import jakarta.ws.rs.WebApplicationException;
@@ -107,6 +108,8 @@ public class RestClientRequestContext extends AbstractResteasyReactiveContext<Re
     private Map<Class<?>, MultipartResponseData> multipartResponsesData;
     private StackTraceElement[] callerStackTrace;
 
+    private final AtomicBoolean userCanceled = new AtomicBoolean();
+
     public RestClientRequestContext(ClientImpl restClient,
             HttpClient httpClient, String httpMethod, URI uri,
             ConfigurationImpl configuration, ClientRequestHeaders requestHeaders,
@@ -601,4 +604,12 @@ private Boolean getBooleanProperty(String name, Boolean defaultValue) {
     protected boolean isRequestScopeManagementRequired() {
         return false;
     }
+
+    public void setUserCanceled() {
+        userCanceled.set(true);
+    }
+
+    public boolean isUserCanceled() {
+        return userCanceled.get();
+    }
 }

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

@@ -50,9 +50,14 @@ public void run() {
                 // be very defensive here as things could have been nulled out when the application is being torn down
                 RestClientRequestContext restClientRequestContext = restClientRequestContextRef.get();
                 if (restClientRequestContext != null) {
+                    restClientRequestContext.setUserCanceled();
+
                     HttpClientRequest httpClientRequest = restClientRequestContext.getHttpClientRequest();
                     if (httpClientRequest != null) {
+                        // if there is already an HTTP request in flight, cancel it
                         httpClientRequest.reset();
+                    } else {
+                        // by having already done setUserCanceled, Quarkus knows to reset the request when it finally gets created
                     }
                 }
             }