GH2701: Fuseki Mod to list and abort running executions.

Author: Aklakan

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

@@ -161,8 +161,10 @@ public static <T> T getOrElseThrow(CompletableFuture<T> cf) {
         } catch (CompletionException ex) {
             if ( ex.getCause() != null ) {
                 Throwable cause = ex.getCause();
-                if ( cause instanceof RuntimeException )
+                if ( cause instanceof RuntimeException ) {
+                    cause.addSuppressed(new RuntimeException("Passed through here."));
                     throw (RuntimeException)cause;
+                }
                 if ( cause instanceof IOException ) {
                     IOException iox = (IOException)cause;
                     // Rather than an HTTP exception, bad authentication becomes IOException("too many authentication attempts");
@@ -173,10 +175,12 @@ public static <T> T getOrElseThrow(CompletableFuture<T> cf) {
                     IO.exception((IOException)cause);
                 }
             }
+            ex.addSuppressed(new RuntimeException("Passed through here."));
             throw ex;
         }
     }
 
+
     /**
      * MUST consume or close the input stream
      * @see HttpLib#finish(HttpResponse)

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;
@@ -557,7 +558,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()));
     }
 
     /**
@@ -577,13 +578,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;
 
@@ -599,29 +622,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);
+            });
     }
 
     /**
@@ -637,25 +647,53 @@ 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));
+        return AsyncHttpRDF.getOrElseThrow(executeJDKAsync(httpClient, httpRequest, bodyHandler));
+    }
+
+    /**
+     * 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;
+            })
+            .exceptionally(ex -> {
+                Throwable cause = ex instanceof CompletionException ? ex.getCause() : ex;
+                if (cause instanceof IOException) {
+                    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));
+                        }
+                    }
+                    // Note: Can't reuse AsyncHttpRDF.handleRuntimeException because of this HttpException.
+                    throw new HttpException(httpRequest.method()+" "+httpRequest.uri().toString(), cause);
+                } else if (cause instanceof RuntimeException re) {
+                    re.addSuppressed(new RuntimeException("Passed through here."));
+                    throw re;
+                } else {
+                    throw new RuntimeException(cause);
                 }
-            }
-            throw new HttpException(httpRequest.method()+" "+httpRequest.uri().toString(), ex);
-        }
+            });
+        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,24 @@
 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 {
+    /** @see #authExecuteAsync */
+    public static <T> HttpResponse<T> authExecute(HttpClient httpClient, HttpRequest httpRequest, BodyHandler<T> bodyHandler) {
+        return AsyncHttpRDF.getOrElseThrow(authExecuteAsync(httpClient, httpRequest, bodyHandler));
+    }
+
     /**
-     * Call {@link HttpClient#send} after applying an active {@link AuthRequestModifier}
+     * 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.
@@ -50,25 +57,26 @@ public class AuthLib {
      * @param bodyHandler
      * @return HttpResponse&lt;T&gt;
      */
-    public static <T> HttpResponse<T> authExecute(HttpClient httpClient, HttpRequest httpRequest, BodyHandler<T> bodyHandler) {
-        HttpResponse<T> httpResponse = HttpLib.executeJDK(httpClient, httpRequest, bodyHandler);
-
-        // -- 401 handling.
-        if ( httpResponse.statusCode() != 401 )
-            return httpResponse;
-        HttpResponse<T> httpResponse2 = handle401(httpClient, httpRequest, bodyHandler, httpResponse);
-        return httpResponse2;
+    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 +110,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 +127,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;
     }

jena-arq/src/main/java/org/apache/jena/http/sys/ExecHTTPBuilder.java

@@ -27,6 +27,7 @@
 import org.apache.jena.query.*;
 import org.apache.jena.sparql.core.Var;
 import org.apache.jena.sparql.engine.binding.Binding;
+import org.apache.jena.sparql.engine.dispatch.SparqlDispatcherRegistry;
 import org.apache.jena.sparql.exec.http.Params;
 import org.apache.jena.sparql.exec.http.QuerySendMode;
 import org.apache.jena.sparql.syntax.syntaxtransform.QueryTransformOps;
@@ -44,7 +45,7 @@ public abstract class ExecHTTPBuilder<X, Y> {
     protected String serviceURL = null;
     private Query query = null;
     protected String queryString = null;
-    protected boolean parseCheck = true;
+    protected Boolean parseCheck = null;
     private HttpClient httpClient = null;
     protected Map<String, String> httpHeaders = new HashMap<>();
     protected Params params = Params.create();
@@ -79,6 +80,10 @@ public Y parseCheck(boolean parseCheck) {
         return thisBuilder();
     }
 
+    protected boolean effectiveParseCheck() {
+        return SparqlDispatcherRegistry.effectiveParseCheck(parseCheck, contextAcc);
+    }
+
     /** Set the query - this also sets the query string to agree with the query argument. */
     public Y query(Query query) {
         Objects.requireNonNull(query);
@@ -93,14 +98,14 @@ public Y query(Query query) {
      */
     public Y query(String queryStr) {
         Objects.requireNonNull(queryStr);
-        Query query = parseCheck ? QueryFactory.create(queryStr) : null;
+        Query query = effectiveParseCheck() ? QueryFactory.create(queryStr) : null;
         setQuery(query, queryStr);
         return thisBuilder();
     }
 
     public Y query(String queryStr, Syntax syntax) {
         Objects.requireNonNull(queryStr);
-        Query query = QueryFactory.create(queryStr, syntax);
+        Query query = effectiveParseCheck() ? QueryFactory.create(queryStr, syntax) : null;
         setQuery(query, queryStr);
         return thisBuilder();
     }