GH-3471: Async HTTP

Author: Aklakan

jena-arq/src/main/java/org/apache/jena/atlas/web/HttpException.java

@@ -34,14 +34,15 @@ public HttpException(int statusCode) {
     }
 
     public HttpException(int statusCode, String statusLine) {
-        super(exMessage(statusCode, statusLine));
-        this.statusCode = statusCode;
-        this.statusLine = statusLine ;
-        this.response = null;
+        this(statusCode, statusLine, null, null);
     }
 
     public HttpException(int statusCode, String statusLine, String responseMessage) {
-        super(exMessage(statusCode, statusLine));
+        this(statusCode, statusLine, responseMessage, null);
+    }
+
+    public HttpException(int statusCode, String statusLine, String responseMessage, Throwable cause) {
+        super(exMessage(statusCode, statusLine), cause);
         this.statusCode = statusCode;
         this.statusLine = statusLine ;
         this.response = responseMessage;

jena-arq/src/main/java/org/apache/jena/http/AsyncHttpRDF.java

@@ -160,26 +160,50 @@ public static void syncOrElseThrow(CompletableFuture<Void> cf) {
      * This operation extracts RuntimeException from the {@code CompletableFuture}.
      */
     public static <T> T getOrElseThrow(CompletableFuture<T> cf) {
+        return getOrElseThrow(cf, null);
+    }
+
+    /**
+     * Get the value of a {@link CompletableFuture} that executes of an HTTP request.
+     * In case on any error, an {@link HttpException} is thrown.
+     *
+     * @param <T> The type of the value being computed.
+     * @param cf The completable future.
+     * @param httpRequest An optional HttpRequest for improving feedback in case of exceptions.
+     * @return The value computed by the completable future.
+     */
+    public static <T> T getOrElseThrow(CompletableFuture<T> cf, HttpRequest httpRequest) {
         Objects.requireNonNull(cf);
         try {
             return cf.join();
         //} catch (CancellationException ex1) { // Let this pass out.
         } catch (CompletionException ex) {
-            if ( ex.getCause() != null ) {
-                Throwable cause = ex.getCause();
-                if ( cause instanceof RuntimeException )
-                    throw (RuntimeException)cause;
+            Throwable cause = ex.getCause();
+            if ( cause != null ) {
+
+                // Pass on our own HttpException instances such as 401 Unauthorized.
+                if ( cause instanceof HttpException httpEx ) {
+                    throw new HttpException(httpEx.getStatusCode(), httpEx.getStatusLine(), httpEx.getResponse(), cause);
+                }
+
+                final String msg = cause.getMessage();
+
                 if ( cause instanceof IOException ) {
-                    IOException iox = (IOException)cause;
                     // Rather than an HTTP exception, bad authentication becomes IOException("too many authentication attempts");
-                    if ( iox.getMessage().contains("too many authentication attempts") ||
-                            iox.getMessage().contains("No credentials provided") ) {
-                        throw new HttpException(401, HttpSC.getMessage(401));
+                    if ( msg != null &&
+                            ( msg.contains("too many authentication attempts") ||
+                              msg.contains("No credentials provided") ) ) {
+                        throw new HttpException(401, HttpSC.getMessage(401), null, cause);
+                    }
+                    if (httpRequest != null) {
+                        throw new HttpException(httpRequest.method()+" "+httpRequest.uri().toString(), cause);
                     }
-                    IO.exception((IOException)cause);
                 }
+
+                throw new HttpException(msg, cause);
             }
-            throw ex;
+            // Note: CompletionException without cause should never happen.
+            throw new HttpException(ex);
         }
     }
 

jena-arq/src/main/java/org/apache/jena/http/HttpLib.java

@@ -37,6 +37,7 @@
 import java.time.Duration;
 import java.util.*;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -560,7 +561,7 @@ public static Builder contentTypeHeader(Builder builder, String contentType) {
      * @return HttpResponse
      */
     public static HttpResponse<InputStream> execute(HttpClient httpClient, HttpRequest httpRequest) {
-        return execute(httpClient, httpRequest, BodyHandlers.ofInputStream());
+        return AsyncHttpRDF.getOrElseThrow(executeAsync(httpClient, httpRequest, BodyHandlers.ofInputStream()), httpRequest);
     }
 
     /**
@@ -580,13 +581,35 @@ public static HttpResponse<InputStream> execute(HttpClient httpClient, HttpReque
      *
      * @param httpClient
      * @param httpRequest
-     * @param bodyHandler
      * @return HttpResponse
      */
-    /*package*/ static <X> HttpResponse<X> execute(HttpClient httpClient, HttpRequest httpRequest, BodyHandler<X> bodyHandler) {
+    public static CompletableFuture<HttpResponse<InputStream>> executeAsync(HttpClient httpClient, HttpRequest httpRequest) {
+        return executeAsync(httpClient, httpRequest, BodyHandlers.ofInputStream());
+    }
+
+    /**
+     * Execute a request, return a {@code HttpResponse<X>} which
+     * can be passed to {@link #handleHttpStatusCode(HttpResponse)} which will
+     * convert non-2xx status code to {@link HttpException HttpExceptions}.
+     * <p>
+     * This function applies the HTTP authentication challenge support
+     * and will repeat the request if necessary with added authentication.
+     * <p>
+     * See {@link AuthEnv} for authentication registration.
+     * <br/>
+     * See {@link #executeJDK} to execute exactly once without challenge response handling.
+     *
+     * @see AuthEnv AuthEnv for authentic registration
+     * @see #executeJDK executeJDK to execute exacly once.
+     *
+     * @param httpClient
+     * @param httpRequest
+     * @param bodyHandler
+     * @return HttpResponse
+     */    /*package*/ static <X> CompletableFuture<HttpResponse<X>> executeAsync(HttpClient httpClient, HttpRequest httpRequest, BodyHandler<X> bodyHandler) {
         // To run with no jena-supplied authentication handling.
         if ( false )
-            return executeJDK(httpClient, httpRequest, bodyHandler);
+            return executeJDKAsync(httpClient, httpRequest, bodyHandler);
         URI uri = httpRequest.uri();
         URI key = null;
 
@@ -602,29 +625,16 @@ public static HttpResponse<InputStream> execute(HttpClient httpClient, HttpReque
                 authEnv.registerUsernamePassword(key, userpasswd[0], userpasswd[1]);
             }
         }
-        try {
-            return AuthLib.authExecute(httpClient, httpRequest, bodyHandler);
-        } finally {
-            if ( key != null )
-                // The AuthEnv is "per tenant".
-                // Temporary registration within the AuthEnv of the
-                // user:password is acceptable.
-                authEnv.unregisterUsernamePassword(key);
-        }
-    }
 
-    /**
-     * Execute request and return a {@code HttpResponse<InputStream>} response.
-     * Status codes have not been handled. The response can be passed to
-     * {@link #handleResponseInputStream(HttpResponse)} which will convert non-2xx
-     * status code to {@link HttpException HttpExceptions}.
-     *
-     * @param httpClient
-     * @param httpRequest
-     * @return HttpResponse
-     */
-    public static HttpResponse<InputStream> executeJDK(HttpClient httpClient, HttpRequest httpRequest) {
-        return execute(httpClient, httpRequest, BodyHandlers.ofInputStream());
+        URI finalKey = key;
+        return AuthLib.authExecuteAsync(httpClient, httpRequest, bodyHandler)
+            .whenComplete((httpResponse, throwable) -> {
+                if ( finalKey != null )
+                    // The AuthEnv is "per tenant".
+                    // Temporary registration within the AuthEnv of the
+                    // user:password is acceptable.
+                    authEnv.unregisterUsernamePassword(finalKey);
+            });
     }
 
     /**
@@ -640,25 +650,32 @@ public static HttpResponse<InputStream> executeJDK(HttpClient httpClient, HttpRe
      * @return HttpResponse
      */
     public static <T> HttpResponse<T> executeJDK(HttpClient httpClient, HttpRequest httpRequest, BodyHandler<T> bodyHandler) {
-        try {
-            // This is the one place all HTTP requests go through.
-            logRequest(httpRequest);
-            HttpResponse<T> httpResponse = httpClient.send(httpRequest, bodyHandler);
-            logResponse(httpResponse);
-            return httpResponse;
-        //} catch (HttpTimeoutException ex) {
-        } catch (IOException | InterruptedException ex) {
-            if ( ex.getMessage() != null ) {
-                // This is silly.
-                // Rather than an HTTP exception, bad authentication becomes IOException("too many authentication attempts");
-                // or IOException("No credentials provided") if the authenticator decides to return null.
-                if ( ex.getMessage().contains("too many authentication attempts") ||
-                     ex.getMessage().contains("No credentials provided") ) {
-                    throw new HttpException(401, HttpSC.getMessage(401));
-                }
-            }
-            throw new HttpException(httpRequest.method()+" "+httpRequest.uri().toString(), ex);
-        }
+        return AsyncHttpRDF.getOrElseThrow(executeJDKAsync(httpClient, httpRequest, bodyHandler), httpRequest);
+    }
+
+    /**
+     * Execute request and return a {@code HttpResponse<InputStream>} response.
+     * Status codes have not been handled. The response can be passed to
+     * {@link #handleResponseInputStream(HttpResponse)} which will convert non-2xx
+     * status code to {@link HttpException HttpExceptions}.
+     *
+     * @param httpClient
+     * @param httpRequest
+     * @return HttpResponse
+     */
+    public static CompletableFuture<HttpResponse<InputStream>> executeJDKAsync(HttpClient httpClient, HttpRequest httpRequest) {
+        return executeAsync(httpClient, httpRequest, BodyHandlers.ofInputStream());
+    }
+
+    public static <T> CompletableFuture<HttpResponse<T>> executeJDKAsync(HttpClient httpClient, HttpRequest httpRequest, BodyHandler<T> bodyHandler) {
+        // This is the one place all HTTP requests go through.
+        logRequest(httpRequest);
+        CompletableFuture<HttpResponse<T>> future = httpClient.sendAsync(httpRequest, bodyHandler)
+            .thenApply(httpResponse -> {
+                logResponse(httpResponse);
+                return httpResponse;
+            });
+        return future;
     }
 
     /*package*/ static CompletableFuture<HttpResponse<InputStream>> asyncExecute(HttpClient httpClient, HttpRequest httpRequest) {

jena-arq/src/main/java/org/apache/jena/http/auth/AuthLib.java

@@ -29,17 +29,19 @@
 import java.nio.charset.StandardCharsets;
 import java.util.Base64;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 
 import org.apache.jena.atlas.lib.Bytes;
 import org.apache.jena.atlas.web.AuthScheme;
 import org.apache.jena.atlas.web.HttpException;
+import org.apache.jena.http.AsyncHttpRDF;
 import org.apache.jena.http.HttpLib;
 import org.apache.jena.riot.web.HttpNames;
 import org.apache.jena.web.HttpSC;
 
 public class AuthLib {
     /**
-     * Call {@link HttpClient#send} after applying an active {@link AuthRequestModifier}
+     * Call the {@link HttpClient} after applying an active {@link AuthRequestModifier}
      * to modify the {@link java.net.http.HttpRequest.Builder}.
      * If no {@link AuthRequestModifier} is available and if a 401 response is received,
      * setup a {@link AuthRequestModifier} passed on registered username and password information.
@@ -51,24 +53,41 @@ public class AuthLib {
      * @return HttpResponse<T>
      */
     public static <T> HttpResponse<T> authExecute(HttpClient httpClient, HttpRequest httpRequest, BodyHandler<T> bodyHandler) {
-        HttpResponse<T> httpResponse = HttpLib.executeJDK(httpClient, httpRequest, bodyHandler);
+        return AsyncHttpRDF.getOrElseThrow(authExecuteAsync(httpClient, httpRequest, bodyHandler), httpRequest);
+    }
 
-        // -- 401 handling.
-        if ( httpResponse.statusCode() != 401 )
-            return httpResponse;
-        HttpResponse<T> httpResponse2 = handle401(httpClient, httpRequest, bodyHandler, httpResponse);
-        return httpResponse2;
+    /**
+     * Call {@link HttpClient#sendAsync} after applying an active {@link AuthRequestModifier}
+     * to modify the {@link java.net.http.HttpRequest.Builder}.
+     * If no {@link AuthRequestModifier} is available and if a 401 response is received,
+     * setup a {@link AuthRequestModifier} passed on registered username and password information.
+     * This function supports basic and digest authentication.
+     *
+     * @param httpClient HttpClient
+     * @param httpRequest
+     * @param bodyHandler
+     * @return CompletableFuture&lt;HttpResponse&lt;T&gt;&gt;
+     */
+    public static <T> CompletableFuture<HttpResponse<T>> authExecuteAsync(HttpClient httpClient, HttpRequest httpRequest, BodyHandler<T> bodyHandler) {
+        return HttpLib.executeJDKAsync(httpClient, httpRequest, bodyHandler)
+            .thenCompose(httpResponse -> {
+                // -- 401 handling.
+                if ( httpResponse.statusCode() != 401 )
+                    return CompletableFuture.completedFuture(httpResponse);
+                CompletableFuture<HttpResponse<T>> httpResponse2 = handle401Async(httpClient, httpRequest, bodyHandler, httpResponse);
+                return httpResponse2;
+            });
     }
 
     /* Handle a 401 (authentication challenge). */
-    private static <T> HttpResponse<T> handle401(HttpClient httpClient,
+    private static <T> CompletableFuture<HttpResponse<T>> handle401Async(HttpClient httpClient,
                                                  HttpRequest request,
                                                  BodyHandler<T> bodyHandler,
                                                  HttpResponse<T> httpResponse401) {
         AuthChallenge aHeader = wwwAuthenticateHeader(httpResponse401);
         if ( aHeader == null )
             // No valid header - simply return the original response.
-            return httpResponse401;
+            return CompletableFuture.completedFuture(httpResponse401);
 
         // Currently on a URI endpoint-by-endpoint basis.
         // String realm = aHeader.getRealm();
@@ -102,14 +121,14 @@ private static <T> HttpResponse<T> handle401(HttpClient httpClient,
             }
             case UNKNOWN :
                 // Not handled. Pass back the 401.
-                return httpResponse401;
+                return CompletableFuture.completedFuture(httpResponse401);
             default:
                 throw new HttpException("Not an authentication scheme -- "+aHeader.authScheme);
         }
 
         // Failed to generate a request modifier for a retry.
         if ( authRequestModifier == null)
-            return httpResponse401;
+            return CompletableFuture.completedFuture(httpResponse401);
 
         // ---- Register for next time the app calls this URI.
         AuthEnv.get().registerAuthModifier(request.uri().toString(), authRequestModifier);
@@ -119,7 +138,7 @@ private static <T> HttpResponse<T> handle401(HttpClient httpClient,
         request2builder = authRequestModifier.addAuth(request2builder);
 
         HttpRequest httpRequest2 = request2builder.build();
-        HttpResponse<T> httpResponse2 = HttpLib.executeJDK(httpClient, httpRequest2, bodyHandler);
+        CompletableFuture<HttpResponse<T>> httpResponse2 = HttpLib.executeJDKAsync(httpClient, httpRequest2, bodyHandler);
         // Pass back to application regardless of response code.
         return httpResponse2;
     }