/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 *
 *   SPDX-License-Identifier: Apache-2.0
 */

package org.apache.jena.http;

import java.io.IOException;
import java.io.InputStream;
import java.io.UncheckedIOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.http.HttpClient;
import java.net.http.HttpHeaders;
import java.net.http.HttpRequest;
import java.net.http.HttpRequest.BodyPublisher;
import java.net.http.HttpRequest.BodyPublishers;
import java.net.http.HttpRequest.Builder;
import java.net.http.HttpResponse;
import java.net.http.HttpResponse.BodyHandler;
import java.net.http.HttpResponse.BodyHandlers;
import java.net.http.HttpResponse.BodySubscribers;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.zip.GZIPInputStream;
import java.util.zip.InflaterInputStream;

import org.apache.commons.compress.compressors.bzip2.BZip2CompressorInputStream;
import org.apache.jena.atlas.RuntimeIOException;
import org.apache.jena.atlas.io.IO;
import org.apache.jena.atlas.lib.IRILib;
import org.apache.jena.atlas.lib.InternalErrorException;
import org.apache.jena.atlas.logging.FmtLog;
import org.apache.jena.atlas.web.HttpException;
import org.apache.jena.atlas.web.TypedInputStream;
import org.apache.jena.http.auth.AuthEnv;
import org.apache.jena.http.auth.AuthLib;
import org.apache.jena.http.sys.HttpRequestModifier;
import org.apache.jena.http.sys.RegistryRequestModifier;
import org.apache.jena.query.ARQ;
import org.apache.jena.riot.web.HttpNames;
import org.apache.jena.sparql.exec.http.Params;
import org.apache.jena.sparql.util.Context;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Operations related to SPARQL HTTP requests - Query, Update and Graph Store protocols.
 * This class is not considered "API".
 */
public class HttpLib {

    // JDK HTTP debug:
    // -Djdk.httpclient.HttpClient.log=errors,requests,headers,frames[:control:data:window:all..],content,ssl,trace,channel

    private HttpLib() {}

    private static Logger LOG = LoggerFactory.getLogger(HttpLib.class.getPackageName()+".HTTP");

    public static BodyHandler<Void> noBody() { return BodyHandlers.discarding(); }

    public static BodyPublisher stringBody(String str) { return BodyPublishers.ofString(str); }

    private static BodyHandler<InputStream> bodyHandlerInputStream = buildDftBodyHandlerInputStream();

    private static BodyHandler<InputStream> buildDftBodyHandlerInputStream() {
        return responseInfo -> {
            return BodySubscribers.ofInputStream();
        };
    }

    /** Read the body of a response as a string in UTF-8. */
    private static Function<HttpResponse<InputStream>, String> bodyInputStreamToString = r-> {
        try {
            InputStream in = r.body();
            String msg = IO.readWholeFileAsUTF8(in);
            return msg;
        } catch (Throwable ex) {
            throw HttpException.builder().cause(ex).build();
        }
    };

    /**
     * Calculate basic auth header value. Use with header "Authorization" (constant
     * {@link HttpNames#hAuthorization}). Best used over https.
     */
    public static String basicAuth(String username, String password) {
        return "Basic " + Base64.getEncoder().encodeToString((username + ":" + password).getBytes(StandardCharsets.UTF_8));
    }

    public static String BEARER = "Bearer";
    public static String BEARER_PREFIX = BEARER+" ";

    /**
     * Calculate bearer auth header value.
     * The token supplied is expected to already be in base 64.
     * Use with header "Authorization" (constant {@link HttpNames#hAuthorization}).
     */
    public static String bearerAuthHeader(String tokenBase64) {
        Objects.requireNonNull(tokenBase64);
        if ( tokenBase64.indexOf(' ') >= 0 )
            throw new IllegalArgumentException("Base64 token contains a space");
        return BEARER_PREFIX + tokenBase64;
    }

    /**
     * Extract the token, without decoding,
     * The token supplied is expected to already be in base 64.
     * Use with header "Authorization" (constant {@link HttpNames#hAuthorization}).
     */
    public static String bearerAuthTokenFromHeader(String authHeaderString) {
        Objects.requireNonNull(authHeaderString);
        if ( ! authHeaderString.startsWith(BEARER_PREFIX) ) {
            throw new IllegalArgumentException("Auth headerString does not start 'Bearer ...'");
        }
        return authHeaderString.substring("Bearer ".length()).trim();
    }

    /**
     * Get the InputStream from an HttpResponse, handling possible compression settings.
     * The application must consume or close the {@code InputStream} (see {@link #finishInputStream(InputStream)}).
     * Closing the InputStream may close the HTTP connection.
     * Assumes the status code has been handled e.g. {@link #handleHttpStatusCode} has been called.
     */
    public static InputStream getInputStream(HttpResponse<InputStream> httpResponse) {
        String encoding = httpResponse.headers().firstValue(HttpNames.hContentEncoding).orElse("");
        InputStream responseInput = httpResponse.body();
        // Only support "Content-Encoding: <compression>" and not
        // "Content-Encoding: chunked, <compression>"
        try {
            switch (encoding) {
                case "" :
                case "identity" : // Proper name for no compression.
                    return responseInput;
                case "gzip" :
                    return new GZIPInputStream(responseInput, 2*1024);
                case "bzip2" :
                    return new BZip2CompressorInputStream(responseInput, true);
                case "inflate" :
                    return new InflaterInputStream(responseInput);
                case "br" : // RFC7932
                default :
                    throw new UnsupportedOperationException("Not supported: Content-Encoding: " + encoding);
            }
        } catch (IOException ex) {
            throw new UncheckedIOException(ex);
        }
    }

    /**
     * Deal with status code and any error message sent as a body in the response.
     * <p>
     * It is this handling 4xx/5xx error messages in the body that forces the use of
     * {@code InputStream}, not generic {@code T}. We don't know until we see the
     * status code how we are going to process the response body.
     * <p>
     * Exits normally without processing the body if the response is 200.
     * <p>
     * Throws {@link HttpException} for 3xx (redirection should have happened by
     * now), 4xx and 5xx, having consumed the body input stream.
     */
    public static void handleHttpStatusCode(HttpResponse<InputStream> response) {
        int httpStatusCode = response.statusCode();
        // There is no status message in HTTP/2.
        if ( ! inRange(httpStatusCode, 100, 599) ) {
            throw HttpException.error("Status code out of range: "+httpStatusCode);
        }
        if ( inRange(httpStatusCode, 100, 199) ) {
            // Informational
            return;
        }
        if ( inRange(httpStatusCode, 200, 299) ) {
            // Success. Continue processing.
            return;
        }
        if ( inRange(httpStatusCode, 300, 399) ) {
            // We had "follow redirects" on (default client) so it's http->https,
            // or the application passed in a HttpClient with redirects off.
            // Either way, we should not continue processing.
            throw exception(response, httpStatusCode);
        }
        if ( inRange(httpStatusCode, 400, 499) ) {
            throw exception(response, httpStatusCode);
        }
        if ( inRange(httpStatusCode, 500, 599) ) {
            throw exception(response, httpStatusCode);
        }

        throw new InternalErrorException("Unknown status code: "+httpStatusCode);
    }

    /**
     * Handle the HTTP response (see {@link #handleHttpStatusCode(HttpResponse)}) and
     * return the InputStream if a 200.
     *
     * @param httpResponse
     * @return InputStream
     */
    public static InputStream handleResponseInputStream(HttpResponse<InputStream> httpResponse) {
        handleHttpStatusCode(httpResponse);
        return getInputStream(httpResponse);
    }

    /**
     * Handle the HTTP response (see {@link #handleHttpStatusCode(HttpResponse)}) and
     * return the TypedInputStream that includes the {@code Content-Type} if a 200.
     *
     * @param httpResponse
     * @return TypedInputStream
     */
    public static TypedInputStream handleResponseTypedInputStream(HttpResponse<InputStream> httpResponse) {
        InputStream input = handleResponseInputStream(httpResponse);
        String ct = HttpLib.responseHeader(httpResponse, HttpNames.hContentType);
        return new TypedInputStream(input, ct);
    }

    /**
     * Handle the HTTP response and consume the body if a 200.
     * Otherwise, throw an {@link HttpException}.
     * @param response
     */
    public static void handleResponseNoBody(HttpResponse<InputStream> response) {
        handleHttpStatusCode(response);
        finishResponse(response);
    }

    /**
     * Handle the HTTP response and read the body to produce a string if a 200.
     * Otherwise, throw an {@link HttpException}.
     * @param response
     * @return String
     */
    public static String handleResponseRtnString(HttpResponse<InputStream> response) {
        return handleResponseRtnString(response, null);
    }

    /**
     * Handle the HTTP response and read the body to produce a string if a 200.
     * Otherwise, throw an {@link HttpException}.
     * @param response
     * @param callback A callback that receives the opened input stream.
     * @return String
     */
    public static String handleResponseRtnString(HttpResponse<InputStream> response, Consumer<InputStream> callback) {
        InputStream input = handleResponseInputStream(response);
        if (callback != null) {
            callback.accept(input);
        }
        try {
            return IO.readWholeFileAsUTF8(input);
        } catch (RuntimeIOException e) {
            throw HttpException.builder().cause(e).build();
        } finally {
            finishInputStream(input);
        }
    }

    /**
     * Clear up and generate an exception - used for 4xx and 5xx, 3xx when not following redirects.
     * This consumes the response body.
     */
    static HttpException exception(HttpResponse<InputStream> response, int httpStatusCode) {
        InputStream in = response.body();
        if ( in == null )
            return HttpException.create(response);
        try {
            String msg;
            try {
                msg = IO.readWholeFileAsUTF8(in);
                if ( msg.isBlank())
                    msg = null;
            } catch (RuntimeIOException e) {
                msg = null;
            }
            return HttpException.builder()
                    .statusCode(httpStatusCode)
                    .responseMessage(msg)
                    .httpResponseHeaders(response.headers())
                    .build();
        } finally { IO.close(in); }
    }

    private static long getContentLength(HttpResponse<InputStream> response) {
        Optional<String> x = response.headers().firstValue(HttpNames.hContentLength);
        if ( x.isEmpty() )
            return -1;
        try {
            return Long.parseLong(x.orElseThrow());
        } catch (NumberFormatException ex) { return -1; }
    }

    /** Test x:int in [min, max] */
    private static boolean inRange(int x, int min, int max) { return min <= x && x <= max; }

    /** Finish with {@code HttpResponse<InputStream>}.
     * This read and drops any remaining bytes in the response body.
     * {@code close} may close the underlying HTTP connection.
     *  See {@link BodySubscribers#ofInputStream()}.
     */
    public static void finishResponse(HttpResponse<InputStream> response) {
        InputStream input = response.body();
        if ( input == null )
            return;
        finishInputStream(input);
    }

    /** Read to end of {@link InputStream}.
     *  {@code close} may close the underlying HTTP connection.
     *  See {@link BodySubscribers#ofInputStream()}.
     */
    public static void finishInputStream(InputStream input) {
        consumeAndClose(input);
    }

    // This is extracted from commons-io, IOUtils.skip.
    // Changes:
    // * No exception.
    // * Always consumes to the end of stream, or the stream throws IOException.
    // This function closes the input stream.

    private static final int SKIP_BUFFER_SIZE = 16*1024;
    // This buffer is only every written. It does not matter if
    /// two threads overlap while writing because we never read the buffer.
    // What matters is that it is fixed size.
    private static final byte[] SKIP_BYTE_BUFFER_WO = new byte[SKIP_BUFFER_SIZE];

    private static void consumeAndClose(final InputStream input) {
        long bytesRead = 0; // Informational
        try {
            for(;;) {
                // See https://issues.apache.org/jira/browse/IO-203 for why we use read() rather than delegating to skip()
                final long n = input.read(SKIP_BYTE_BUFFER_WO, 0, SKIP_BUFFER_SIZE);
                if (n < 0) { // EOF
                    break;
                }
                bytesRead += n;
            }
        } catch (IOException ex) { /*ignore*/ }
        try { input.close() ; } catch (Throwable th) { /*silent*/ }
    }

    /** String to {@link URI}. Throws {@link HttpException} on bad syntax or if the URI isn't absolute. */
    public static URI toRequestURI(String uriStr) {
        try {
            URI uri = new URI(uriStr);
            if ( ! uri.isAbsolute() )
                throw HttpException.error("Not an absolute URL: <"+uriStr+">");
            return uri;
        } catch (URISyntaxException ex) {
            int idx = ex.getIndex();
            String msg = (idx<0)
                ? String.format("Bad URL: %s", uriStr)
                : String.format("Bad URL: %s starting at character %d", uriStr, idx);
            throw HttpException.error(msg, ex);
        }
    }

    // Terminology:
    // RFC 2616:   Request-Line   = Method SP Request-URI SP HTTP-Version CRLF
    // RFC 7320:   request-line   = method SP request-target SP HTTP-version CRLF
    // https://datatracker.ietf.org/doc/html/rfc7230#section-3.1.1

    // EndpointURI: URL for a service, no query string.

    /** Test whether a URI is a service endpoint. It must be absolute, with host and path, and without query string or fragment. */
    public static boolean isEndpoint(URI uri) {
        return uri.isAbsolute() &&
                uri.getHost() != null &&
                uri.getRawPath() != null &&
                uri.getRawQuery() == null &&
                uri.getRawFragment() == null;
    }

    /**
     * Return a string (assumed to be an absolute URI) without query string or fragment.
     */
    public static String endpoint(String uriStr) {
        int idx1 = uriStr.indexOf('?');
        // Assuming a well-formed URI string, it is path-query-fragment
        // so if there is a query string, don't need to look for a fragment.
        int idx2 = (idx1<0) ? uriStr.indexOf('#') : -1;

        if ( idx1 < 0 && idx2 < 0 )
            return uriStr;

        int idx = -1;
        if ( idx1 < 0 && idx2 > 0 )
            idx = idx2;
        else if ( idx1 > 0 && idx2 < 0 )
            idx = idx1;
        else
            idx = Math.min(idx1,  idx2);
        return uriStr.substring(0, idx);
    }

    /** URI, without query string and fragment. */
    public static URI endpointURI(URI uri) {
        if ( uri.getRawQuery() == null && uri.getRawFragment() == null )
            return uri;
        try {
            // Same URI components except without query string and fragment.
            return new URI(uri.getScheme(), uri.getRawAuthority(), uri.getRawPath(), null, null);
        } catch (URISyntaxException x) {
            throw new IllegalArgumentException(x.getMessage(), x);
        }
    }

    /**
     * The "request target" for digest auth. The server-side name of a resource - no
     * authority (the host part).
     * <p>
     * RFC 7616 (digest auth), section 3.4 The Effective Request URI (Section 5.5 of RFC7230).
     * <p>
     * For SPARQL, the target is the service, not a resource
     * named by the uri+query string.
     * <p>
     * This makes query-by-GET and query-by-POST work the same way.
     */
    public static String requestTargetServer(URI uri) {
        // RFC 7230 5.5
        //   If the request-target is in authority-form or asterisk-form, the
        //   effective request URI's combined path and query component is
        //   empty.
        String path = uri.getRawPath();
        if ( path == null || path.isEmpty() )
            path = "/";
        String target = path;
        String qs = uri.getRawQuery();
        if ( qs != null )
            target = target+"?"+qs;
        return target;
    }

    /** Return a HttpRequest */
    public static HttpRequest newGetRequest(String url, Consumer<HttpRequest.Builder> modifier) {
        HttpRequest.Builder builder = HttpLib.requestBuilderFor(url).uri(toRequestURI(url)).GET();
        if ( modifier != null )
            modifier.accept(builder);
        return builder.build();
    }

    public static <X> X dft(X value, X dftValue) {
        return (value != null) ? value : dftValue;
    }

    public static <X> List<X> copyArray(List<X> array) {
        if ( array == null )
            return null;
        return new ArrayList<>(array);
    }

    /** Encode a string suitable for use in an URL query string */
    public static String urlEncodeQueryString(String str) {
        // java.net.URLEncoder is excessive - it encodes / and : which
        // is not necessary in a query string or fragment.
        String x1 = IRILib.encodeUriQueryFrag(str);
        String x2 = IRILib.encodeNonASCII(x1);
        return x2;
    }

    /** Query string is assumed to already be encoded. */
    public static String requestURL(String url, String queryString) {
        if ( queryString == null || queryString.isEmpty() )
            // Empty string. Don't add "?"
            return url;
        String sep = url.contains("?") ? "&" : "?";
        String requestURL = url+sep+queryString;
        return requestURL;
    }

    public static HttpRequest.Builder requestBuilderFor(String serviceEndpoint) {
        HttpRequest.Builder requestBuilder = HttpRequest.newBuilder();
        return AuthEnv.get().addAuth(requestBuilder, serviceEndpoint);
    }

    public static Builder requestBuilder(String url, Map<String, String> httpHeaders, long readTimeout, TimeUnit readTimeoutUnit) {
        HttpRequest.Builder builder = HttpLib.requestBuilderFor(url);
        headers(builder, httpHeaders);
        builder.uri(toRequestURI(url));
        if ( readTimeout >= 0 )
            builder.timeout(Duration.ofMillis(readTimeoutUnit.toMillis(readTimeout)));
        return builder;
    }

    /** Create a {@code HttpRequest.Builder} from an {@code HttpRequest}. */
    public static HttpRequest.Builder createBuilder(HttpRequest request) {
        HttpRequest.Builder builder = HttpRequest.newBuilder()
                .expectContinue(request.expectContinue())
                .uri(request.uri());
        builder.method(request.method(), request.bodyPublisher().orElse(BodyPublishers.noBody()));
        request.timeout().ifPresent(builder::timeout);
        request.version().ifPresent(builder::version);
        request.headers().map().forEach((name, values)->values.forEach(value->builder.header(name, value)));
        return builder;
    }

    /** Set the headers from the Map if the map is not null. Returns the Builder. */
    static Builder headers(Builder builder, Map<String, String> httpHeaders) {
        if ( httpHeaders != null )
            httpHeaders.forEach(builder::header);
        return builder;
    }

    /** Set the "Accept" header if value is not null. Returns the builder. */
    public static Builder acceptHeader(Builder builder, String acceptHeader) {
        if ( acceptHeader != null )
            builder.header(HttpNames.hAccept, acceptHeader);
        return builder;
    }

    /** Set the "Content-Type" header if value is not null. Returns the builder. */
    public static Builder contentTypeHeader(Builder builder, String contentType) {
        if ( contentType != null )
            builder.header(HttpNames.hContentType, contentType);
        return builder;
    }

    // Disabled. Don't encourage using compression ("Content-Encoding: gzip") because it interacts with streaming.
    // Specifically, streaming (unknown Content-Length) needs chunking. Both chunking and compression
    // encodings use the same HTTP header. Yet they are handled by different layers.
    // The basic http code handles chunking.
    // The complete encoding header can get removed resulting in a compressed stream
    // without any indication being passed to the application.
//    /**
//     * Set the "Accept-Encoding" header. Returns the builder.
//     * See {@link #getInputStream(HttpResponse)}.
//     */
//    public
//    /*package*/ static Builder acceptEncodingCompressed(Builder builder) {
//        builder.header(HttpNames.hAcceptEncoding, WebContent.acceptEncodingCompressed);
//        return builder;
//    }

    /**
     * Execute a request, return a {@code HttpResponse<InputStream>} which
     * can be passed to {@link #handleResponseInputStream(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
     * @return HttpResponse
     */
    public static HttpResponse<InputStream> execute(HttpClient httpClient, HttpRequest httpRequest) {
        return AsyncHttpRDF.getOrElseThrow(executeAsync(httpClient, httpRequest, BodyHandlers.ofInputStream()), httpRequest);
    }

    /**
     * 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
     * @return HttpResponse
     */
    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 executeJDKAsync(httpClient, httpRequest, bodyHandler);
        URI uri = httpRequest.uri();
        URI key = null;

        AuthEnv authEnv = AuthEnv.get();

        if ( uri.getUserInfo() != null ) {
            String[] userpasswd = uri.getUserInfo().split(":");
            if ( userpasswd.length == 2 ) {
                // User info in the URI is not a good idea.
                // Only if "user:password@host", not "user@host"
                key = HttpLib.endpointURI(uri);
                // The auth key will include user:password making it specific.
                authEnv.registerUsernamePassword(key, userpasswd[0], userpasswd[1]);
            }
        }

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

    /**
     * Execute request and return a response without authentication challenge
     * handling. Status codes have not been handled. This is a call to
     * {@code HttpClient.send} converting exceptions to {@link HttpException}.
     * request and responses are logged as "debug" to logger
     * {@code org.apache.jena.http.HTTP}.
     *
     * @param httpClient
     * @param httpRequest
     * @param bodyHandler
     * @return HttpResponse
     */
    public static <T> HttpResponse<T> executeJDK(HttpClient httpClient, HttpRequest httpRequest, BodyHandler<T> bodyHandler) {
        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) {
        logAsyncRequest(httpRequest);
        return httpClient.sendAsync(httpRequest, BodyHandlers.ofInputStream());
    }

    /**
     * Push data. POST, PUT, PATCH request with no response body data.
     * @deprecated Use {@link #httpPushData(HttpClient, HttpMethod, String, Consumer, BodyPublisher)}
     */
    @Deprecated(forRemoval = true)
    public static void httpPushData(HttpClient httpClient, @SuppressWarnings("removal") Push style, String url, Consumer<HttpRequest.Builder> modifier, BodyPublisher body) {
        HttpResponse<InputStream> response = httpPushWithResponse(httpClient, style.method(), url, modifier, body);
        handleResponseNoBody(response);
    }

    /** Push data. POST, PUT, PATCH request with no response body data. */
    public static void httpPushData(HttpClient httpClient, HttpMethod method, String url, Consumer<HttpRequest.Builder> modifier, BodyPublisher body) {
        HttpResponse<InputStream> response = httpPushWithResponse(httpClient, method, url, modifier, body);
        handleResponseNoBody(response);
    }

    /**
     * Push data. POST, PUT, PATCH request with no response body data.
     * @deprecated Use {@link #httpPushWithResponse(HttpClient, HttpMethod, String, Consumer, BodyPublisher)}
     */
    @Deprecated(forRemoval = true)
    public static HttpResponse<InputStream> httpPushWithResponse(HttpClient httpClient, @SuppressWarnings("removal") Push style, String url,
                                                                 Consumer<HttpRequest.Builder> modifier, BodyPublisher body) {
        return httpPushWithResponse(httpClient, style.method(), url, modifier, body);

    }
    // Worker
    public static HttpResponse<InputStream> httpPushWithResponse(HttpClient httpClient, HttpMethod method, String url,
                                                                 Consumer<HttpRequest.Builder> modifier, BodyPublisher body) {
        switch(method) {
            case POST,PUT, PATCH -> {}
            default -> { throw HttpException.error("Method not suitable for push operation: "+method); }
        }
        return httpPushWithResponse(httpClient, method.method(), url, modifier, body);
    }


    private static HttpResponse<InputStream> httpPushWithResponse(HttpClient httpClient, String method, String url,
                                                                  Consumer<HttpRequest.Builder> modifier, BodyPublisher body) {
        URI uri = toRequestURI(url);
        HttpRequest.Builder builder = requestBuilderFor(url);
        builder.uri(uri);
        builder.method(method, body);
        if ( modifier != null )
            modifier.accept(builder);
        HttpResponse<InputStream> response = execute(httpClient, builder.build());
        return response;
    }

    /** Request */
    private static void logRequest(HttpRequest httpRequest) {
        if ( LOG.isDebugEnabled() ) {
            FmtLog.debug(LOG, "> %s %s", httpRequest.method(), httpRequest.uri());
            logHeaders(LOG, httpRequest.headers());
        }
    }

    /** Async Request */
    private static void logAsyncRequest(HttpRequest httpRequest) {}

        /** Response (do not touch the body!)  */
    private static void logResponse(HttpResponse<?> httpResponse) {
        if ( LOG.isDebugEnabled() ) {
            FmtLog.debug(LOG, "< %d %s %s", httpResponse.statusCode(), httpResponse.request().method(), httpResponse.uri());
            logHeaders(LOG,  httpResponse.headers());
//        httpResponse.previousResponse();
        }
    }

    private static void logHeaders(Logger log, HttpHeaders headers) {
        headers.map().forEach((header, values)->{
            values.forEach(value->FmtLog.debug(log, "  %-15s %s", header, value));
        });
    }

    /**
     * Allow setting additional/optional HTTP headers and query parameters on a per remote service (including for SERVICE) basis.
     * <ul>
     * <li>ARQ.httpRequestModifer - the specific modifier</li>
     * <li>ARQ.httpRegistryRequestModifer - the registry, keyed by service URL.</li>
     * </ul>
     */
    public static void modifyByService(String serviceURI, Context context, Params params, Map<String, String> httpHeaders) {
        HttpRequestModifier modifier = context.get(ARQ.httpRequestModifer);
        if ( modifier != null ) {
            modifier.modify(params, httpHeaders);
            return;
        }
        RegistryRequestModifier modifierRegistry = context.get(ARQ.httpRegistryRequestModifer);
        if ( modifierRegistry == null )
            modifierRegistry = RegistryRequestModifier.get();
        if ( modifierRegistry != null ) {
            HttpRequestModifier mods = modifierRegistry.find(serviceURI);
            if ( mods != null )
                mods.modify(params, httpHeaders);
        }
    }

    /**
     * Return a modifier that will set headers using the values in the map.
     * An argument of "null" means "no action".
     */
    public static Consumer<HttpRequest.Builder> setHeaders(Map<String, String> headers) {
        if ( headers == null )
            return (x)->{};
        return x->headers.forEach(x::header);
    }

    /**
     * Return a modifier that will set the Accept header to the value.
     * An argument of "null" means "no action".
     */
    static Consumer<HttpRequest.Builder> setAcceptHeader(String acceptHeader) {
        if ( acceptHeader == null )
            return (x)->{};
        return header(HttpNames.hAccept, acceptHeader);
    }

    /**
     * Return a modifier that will set the Content-Type header to the value.
     * An argument of "null" means "no action".
     */
    static Consumer<HttpRequest.Builder> setContentTypeHeader(String contentType) {
        if ( contentType == null )
            return (x)->{};
        return header(HttpNames.hContentType, contentType);
    }

    /**
     * Return a modifier that will set the named header to the value.
     */
    static Consumer<HttpRequest.Builder> header(String headerName, String headerValue) {
        return x->x.header(headerName, headerValue);
    }

    /** Return the first header of the given name, or null if none */
    public static String responseHeader(HttpResponse<?> response, String headerName) {
        Objects.requireNonNull(response);
        Objects.requireNonNull(headerName);
        return response.headers().firstValue(headerName).orElse(null);
    }

    // HTTP response header inserted to aid tracking.
    public static String FusekiRequestIdHeader = "Fuseki-Request-Id";

    /**
     * Test whether a URL identifies a Fuseki server. This operation can not guarantee to
     * detect a Fuseki server - for example, it may be behind a reverse proxy that masks
     * the signature.
     */
    public static boolean isFuseki(String datasetURL) {
        HttpRequest.Builder builder =
                HttpRequest.newBuilder().uri(toRequestURI(datasetURL)).method(HttpMethod.METHOD_HEAD, BodyPublishers.noBody());
        HttpRequest request = builder.build();
        HttpClient httpClient = HttpEnv.getHttpClient(datasetURL);
        HttpResponse<InputStream> response = execute(httpClient, request);
        handleResponseNoBody(response);

        Optional<String> value1 = response.headers().firstValue(FusekiRequestIdHeader);
        if ( value1.isPresent() )
            return true;
        Optional<String> value2 = response.headers().firstValue("Server");
        if ( value2.isEmpty() )
            return false;
        String headerValue = value2.orElseThrow();
        boolean isFuseki = headerValue.startsWith("Apache Jena Fuseki") ||
                           headerValue.toLowerCase().contains("fuseki");
        return isFuseki;
    }
}
