Expose terminal error as a CompletionStage that completes with the failure cause or null on success.

Author: arturobernalg

httpcore5-reactive/src/main/java/org/apache/hc/core5/reactive/ReactiveResponseConsumer.java

@@ -31,6 +31,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.Future;
@@ -67,6 +68,12 @@ public final class ReactiveResponseConsumer implements AsyncResponseConsumer<Voi
     private final CompletableFuture<Message<HttpResponse, Publisher<ByteBuffer>>> responseCompletableFuture;
     private final CompletableFuture<Void> responseCompletionFuture;
 
+    /**
+     * Completes with {@code null} on success and with the terminal {@link Throwable} on failure.
+     * This future never completes exceptionally.
+     */
+    private final CompletableFuture<Throwable> failureFuture;
+
     private volatile BasicFuture<Void> responseCompletion;
     private volatile HttpResponse informationResponse;
     private volatile EntityDetails entityDetails;
@@ -78,6 +85,7 @@ public ReactiveResponseConsumer() {
         this.responseFuture = new BasicFuture<>(null);
         this.responseCompletableFuture = new CompletableFuture<>();
         this.responseCompletionFuture = new CompletableFuture<>();
+        this.failureFuture = new CompletableFuture<>();
     }
 
     /**
@@ -90,6 +98,7 @@ public ReactiveResponseConsumer(final FutureCallback<Message<HttpResponse, Publi
         this.responseFuture = new BasicFuture<>(Args.notNull(responseCallback, "responseCallback"));
         this.responseCompletableFuture = new CompletableFuture<>();
         this.responseCompletionFuture = new CompletableFuture<>();
+        this.failureFuture = new CompletableFuture<>();
     }
 
     public Future<Message<HttpResponse, Publisher<ByteBuffer>>> getResponseFuture() {
@@ -136,6 +145,16 @@ public CompletionStage<Void> getResponseCompletionStage() {
         return responseCompletionFuture;
     }
 
+    /**
+     * Completes with {@code null} on success and with the terminal {@link Throwable} on failure.
+     * This stage never completes exceptionally.
+     *
+     * @since 5.5
+     */
+    public CompletionStage<Throwable> getFailureStage() {
+        return failureFuture;
+    }
+
     /**
      * Returns the intermediate (1xx) HTTP response if one was received.
      *
@@ -165,10 +184,10 @@ public List<Header> getTrailers() {
 
     @Override
     public void consumeResponse(
-            final HttpResponse response,
-            final EntityDetails entityDetails,
-            final HttpContext httpContext,
-            final FutureCallback<Void> resultCallback
+        final HttpResponse response,
+        final EntityDetails entityDetails,
+        final HttpContext httpContext,
+        final FutureCallback<Void> resultCallback
     ) {
         this.entityDetails = entityDetails;
         this.responseCompletion = new BasicFuture<>(resultCallback);
@@ -190,12 +209,13 @@ public void informationResponse(final HttpResponse response, final HttpContext h
     @Override
     public void failed(final Exception cause) {
         reactiveDataConsumer.failed(cause);
-
-        // Complete stage/futures regardless of whether consumeResponse() has been invoked yet.
         responseFuture.failed(cause);
         responseCompletableFuture.completeExceptionally(cause);
         responseCompletionFuture.completeExceptionally(cause);
 
+        // Record failure as a normal completion value.
+        failureFuture.complete(cause);
+
         final BasicFuture<Void> completion = responseCompletion;
         if (completion != null) {
             completion.failed(cause);
@@ -222,6 +242,9 @@ public void streamEnd(final List<? extends Header> trailers) {
         // Complete CF before BasicFuture.completed(...) (it may trigger releaseResources()).
         responseCompletionFuture.complete(null);
 
+        // Success => no failure.
+        failureFuture.complete(null);
+
         final BasicFuture<Void> completion = responseCompletion;
         if (completion != null) {
             completion.completed(null);
@@ -241,6 +264,10 @@ public void releaseResources() {
             responseCompletionFuture.cancel(true);
         }
 
+        if (!failureFuture.isDone()) {
+            failureFuture.complete(new CancellationException());
+        }
+
         final BasicFuture<Void> completion = responseCompletion;
         if (completion != null) {
             completion.cancel();

httpcore5-reactive/src/test/java/org/apache/hc/core5/reactive/examples/ReactiveClientCompletionStageExample.java

@@ -39,8 +39,6 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
-import io.reactivex.rxjava3.core.Flowable;
-import io.reactivex.rxjava3.core.Observable;
 import org.apache.hc.core5.http.ContentType;
 import org.apache.hc.core5.http.Header;
 import org.apache.hc.core5.http.HttpConnection;
@@ -60,6 +58,9 @@
 import org.apache.hc.core5.util.Timeout;
 import org.reactivestreams.Publisher;
 
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Observable;
+
 /**
  * Client demo using CompletionStage accessors on ReactiveResponseConsumer (Java 8).
  */
@@ -190,6 +191,14 @@ public void onExchangeComplete(final HttpConnection connection, final boolean ke
             final ReactiveResponseConsumer consumer = new ReactiveResponseConsumer();
             requester.execute(requestProducer, consumer, Timeout.ofSeconds(30), null);
 
+            consumer.getFailureStage().whenComplete((t, ex) -> {
+                if (ex != null) {
+                    ex.printStackTrace();
+                } else if (t != null) {
+                    System.out.println("Request failed: " + t);
+                }
+            });
+
             final CompletableFuture<Void> printedAndDrained = new CompletableFuture<>();
 
             consumer.getResponseStage().whenComplete((msg, ex) -> {