robtimus · September 7, 2025 12:14
diff --git a/HttpClientConnection.java b/HttpClientConnection.java
 /*
 * HttpClientConnection.java
 * Copyright 2025 Rob Spoor
 *
 * Licensed 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
 *
 *     http://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.
 */

 package com.github.robtimus.connect.sdk.httpclient;

 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.UncheckedIOException;
 import java.lang.invoke.MethodHandle;
 import java.lang.invoke.MethodHandles;
 import java.lang.invoke.MethodType;
 import java.net.InetSocketAddress;
 import java.net.ProxySelector;
 import java.net.URI;
 import java.net.http.HttpClient;
 import java.net.http.HttpHeaders;
 import java.net.http.HttpRequest;
 import java.net.http.HttpResponse;
 import java.nio.ByteBuffer;
 import java.nio.channels.Channels;
 import java.nio.channels.WritableByteChannel;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Base64;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.Flow.Subscriber;
 import java.util.concurrent.Flow.Subscription;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLParameters;
 import com.worldline.connect.sdk.java.CommunicatorConfiguration;
 import com.worldline.connect.sdk.java.ProxyConfiguration;
 import com.worldline.connect.sdk.java.communication.CommunicationException;
 import com.worldline.connect.sdk.java.communication.Connection;
 import com.worldline.connect.sdk.java.communication.MultipartFormDataObject;
 import com.worldline.connect.sdk.java.communication.PooledConnection;
 import com.worldline.connect.sdk.java.communication.RequestHeader;
 import com.worldline.connect.sdk.java.communication.ResponseHandler;
 import com.worldline.connect.sdk.java.communication.ResponseHeader;
 import com.worldline.connect.sdk.java.domain.UploadableFile;
 import com.worldline.connect.sdk.java.logging.BodyObfuscator;
 import com.worldline.connect.sdk.java.logging.CommunicatorLogger;
 import com.worldline.connect.sdk.java.logging.HeaderObfuscator;
 import com.worldline.connect.sdk.java.logging.LogMessageBuilder;
 import com.worldline.connect.sdk.java.logging.RequestLogMessageBuilder;
 import com.worldline.connect.sdk.java.logging.ResponseLogMessageBuilder;

 /**
 * A {@link Connection} implementation based on {@link HttpClient}.
 *
 * @author Rob Spoor
 */
 @SuppressWarnings("nls")
 public final class HttpClientConnection implements PooledConnection {

    private static final Charset CHARSET = StandardCharsets.UTF_8;

    private static final String PROXY_AUTHORIZATION_HEADER = "Proxy-Authorization";

    // HttpClient only has close and shutdown methods since Java 21, so let's use the first if it's available
    private static final MethodHandle HTTP_CLIENT_CLOSE_HANDLE;

    static {
        MethodHandle closeHandle;
        try {
            closeHandle = MethodHandles.lookup().findVirtual(HttpClient.class, "close", MethodType.methodType(void.class));
        } catch (@SuppressWarnings("unused") NoSuchMethodException e) {
            closeHandle = null;
        } catch (IllegalAccessException e) {
            throw new IllegalStateException(e);
        }
        HTTP_CLIENT_CLOSE_HANDLE = closeHandle;
    }

    private final HttpClient httpClient;

    private final Duration requestTimeout;

    private final String proxyAuthorization;

    private final AtomicReference<BodyObfuscator> bodyObfuscator = new AtomicReference<>(BodyObfuscator.defaultObfuscator());
    private final AtomicReference<HeaderObfuscator> headerObfuscator = new AtomicReference<>(HeaderObfuscator.defaultObfuscator());

    private final AtomicReference<CommunicatorLogger> communicatorLogger = new AtomicReference<>();

    private HttpClientConnection(Builder builder) {
        httpClient = createHttpClient(builder);
        requestTimeout = Duration.ofMillis(builder.socketTimeout);
        proxyAuthorization = createProxyAuthorization(builder);
    }

    private HttpClient createHttpClient(Builder builder) {
        HttpClient.Builder httpClientBuilder = HttpClient.newBuilder()
                .connectTimeout(Duration.ofMillis(builder.connectTimeout));

        ProxyConfiguration proxyConfiguration = builder.proxyConfiguration;
        if (proxyConfiguration != null) {
            InetSocketAddress proxyAddress = InetSocketAddress.createUnresolved(proxyConfiguration.getHost(), proxyConfiguration.getPort());
            ProxySelector proxySelector = ProxySelector.of(proxyAddress);
            httpClientBuilder.proxy(proxySelector);
        }
        if (builder.sslContext != null) {
            httpClientBuilder.sslContext(builder.sslContext);
        }
        if (builder.sslParameters != null) {
            httpClientBuilder.sslParameters(builder.sslParameters);
        }
        return httpClientBuilder.build();
    }

    private String createProxyAuthorization(Builder builder) {
        ProxyConfiguration proxyConfiguration = builder.proxyConfiguration;
        if (proxyConfiguration == null || proxyConfiguration.getUsername() == null) {
            return null;
        }
        String credentials = proxyConfiguration.getUsername() + ":" + Objects.toString(proxyConfiguration.getPassword(), "");
        return "BASIC " + Base64.getEncoder().encodeToString(credentials.getBytes(CHARSET));
    }

    @Override
    public void close() throws IOException {
        if (HTTP_CLIENT_CLOSE_HANDLE != null) {
            try {
                HTTP_CLIENT_CLOSE_HANDLE.invokeExact(httpClient);
            } catch (IOException | RuntimeException e) {
                throw e;
            } catch (Throwable e) {
                throw new IllegalStateException(e);
            }
        }
    }

    @Override
    public <R> R get(URI uri, List<RequestHeader> requestHeaders, ResponseHandler<R> responseHandler) {
        HttpRequest.Builder getBuilder = HttpRequest.newBuilder(uri)
                .GET()
                .timeout(requestTimeout);
        addHeaders(getBuilder, requestHeaders);
        return executeRequest(getBuilder.build(), responseHandler);
    }

    @Override
    public <R> R delete(URI uri, List<RequestHeader> requestHeaders, ResponseHandler<R> responseHandler) {
        HttpRequest.Builder deleteBuilder = HttpRequest.newBuilder(uri)
                .DELETE()
                .timeout(requestTimeout);
        addHeaders(deleteBuilder, requestHeaders);
        return executeRequest(deleteBuilder.build(), responseHandler);
    }

    @Override
    public <R> R post(URI uri, List<RequestHeader> requestHeaders, String body, ResponseHandler<R> responseHandler) {
        HttpRequest.BodyPublisher bodyPublisher = createBodyPublisher(body);
        return post(uri, requestHeaders, bodyPublisher, responseHandler);
    }

    @Override
    public <R> R post(URI uri, List<RequestHeader> requestHeaders, MultipartFormDataObject multipart, ResponseHandler<R> responseHandler) {
        HttpRequest.BodyPublisher bodyPublisher = createBodyPublisher(multipart);
        return post(uri, requestHeaders, bodyPublisher, responseHandler);
    }

    private <R> R post(URI uri, List<RequestHeader> requestHeaders, HttpRequest.BodyPublisher bodyPublisher, ResponseHandler<R> responseHandler) {
        HttpRequest.Builder postBuilder = HttpRequest.newBuilder(uri)
                .POST(bodyPublisher)
                .timeout(requestTimeout);
        addHeaders(postBuilder, requestHeaders);
        return executeRequest(postBuilder.build(), responseHandler);
    }

    @Override
    public <R> R put(URI uri, List<RequestHeader> requestHeaders, String body, ResponseHandler<R> responseHandler) {
        HttpRequest.BodyPublisher bodyPublisher = createBodyPublisher(body);
        return put(uri, requestHeaders, bodyPublisher, responseHandler);
    }

    @Override
    public <R> R put(URI uri, List<RequestHeader> requestHeaders, MultipartFormDataObject multipart, ResponseHandler<R> responseHandler) {
        HttpRequest.BodyPublisher bodyPublisher = createBodyPublisher(multipart);
        return put(uri, requestHeaders, bodyPublisher, responseHandler);
    }

    private <R> R put(URI uri, List<RequestHeader> requestHeaders, HttpRequest.BodyPublisher bodyPublisher, ResponseHandler<R> responseHandler) {
        HttpRequest.Builder putBuilder = HttpRequest.newBuilder(uri)
                .PUT(bodyPublisher)
                .timeout(requestTimeout);
        addHeaders(putBuilder, requestHeaders);
        return executeRequest(putBuilder.build(), responseHandler);
    }

    private static HttpRequest.BodyPublisher createBodyPublisher(String body) {
        return body != null ? new JsonRequestBodyPublisher(body, CHARSET) : HttpRequest.BodyPublishers.noBody();
    }

    private static HttpRequest.BodyPublisher createBodyPublisher(MultipartFormDataObject multipart) {
        return new MultipartBodyPublisher(multipart);
    }

    private <R> R executeRequest(HttpRequest request, ResponseHandler<R> responseHandler) {
        String requestId = UUID.randomUUID().toString();
        logRequest(request, requestId);

        long startTime = System.currentTimeMillis();

        boolean[] logRuntimeExceptions = { true };

        try {
            HttpResponse<R> response = httpClient.send(request, responseInfo -> {
                int statusCode = responseInfo.statusCode();
                List<ResponseHeader> headers = getHeaders(responseInfo);

                HttpResponse.BodySubscriber<InputStream> inputStreamSubscriber = HttpResponse.BodySubscribers.ofInputStream();
                HttpResponse.BodySubscriber<R> subscriber = HttpResponse.BodySubscribers.mapping(inputStreamSubscriber, inputStream -> {

                    // do not log runtime exceptions that originate from the response handler, as those are not communication errors
                    logRuntimeExceptions[0] = false;

                    return responseHandler.handleResponse(statusCode, inputStream, headers);
                });

                CommunicatorLogger logger = communicatorLogger.get();
                if (logger != null) {
                    subscriber = new LoggingResponseBodySubscriber<>(subscriber, requestId, startTime, responseInfo, logger);
                }
                return subscriber;
            });
            return response.body();
        } catch (InterruptedException e) {
            // Restore interrupt status
            Thread.currentThread().interrupt();
            logError(requestId, e, startTime, communicatorLogger.get());
            throw new CommunicationException(e);
        } catch (IOException e) {
            logError(requestId, e, startTime, communicatorLogger.get());
            throw new CommunicationException(e);
        } catch (CommunicationException e) {
            logError(requestId, e, startTime, communicatorLogger.get());
            throw e;
        } catch (RuntimeException e) {
            if (logRuntimeExceptions[0]) {
                logError(requestId, e, startTime, communicatorLogger.get());
            }
            throw e;
        }
    }

    private void addHeaders(HttpRequest.Builder httpRequestBuilder, List<RequestHeader> requestHeaders) {
        if (requestHeaders != null) {
            for (RequestHeader requestHeader : requestHeaders) {
                httpRequestBuilder.header(requestHeader.getName(), requestHeader.getValue());
            }
        }
        if (proxyAuthorization != null) {
            httpRequestBuilder.header(PROXY_AUTHORIZATION_HEADER, proxyAuthorization);
        }
    }

    private List<ResponseHeader> getHeaders(HttpResponse.ResponseInfo httpResponseInfo) {
        Map<String, List<String>> headerMap = httpResponseInfo.headers().map();
        // Under normal circumstances, each header only occurs once
        List<ResponseHeader> result = new ArrayList<>(headerMap.size());
        for (Map.Entry<String, List<String>> entry : headerMap.entrySet()) {
            String headerName = entry.getKey();
            for (String headerValue : entry.getValue()) {
                result.add(new ResponseHeader(headerName, headerValue));
            }
        }
        return result;
    }

    @Override
    public void closeIdleConnections(long idleTime, TimeUnit timeUnit) {
        // HttpClient does not support closing idle connections explicitly
    }

    @Override
    public void closeExpiredConnections() {
        // HttpClient does not support closing expired connections explicitly
    }

    @Override
    public void setBodyObfuscator(BodyObfuscator bodyObfuscator) {
        if (bodyObfuscator == null) {
            throw new IllegalArgumentException("bodyObfuscator is required");
        }
        this.bodyObfuscator.set(bodyObfuscator);
    }

    @Override
    public void setHeaderObfuscator(HeaderObfuscator headerObfuscator) {
        if (headerObfuscator == null) {
            throw new IllegalArgumentException("headerObfuscator is required");
        }
        this.headerObfuscator.set(headerObfuscator);
    }

    @Override
    public void enableLogging(CommunicatorLogger communicatorLogger) {
        if (communicatorLogger == null) {
            throw new IllegalArgumentException("communicatorLogger is required");
        }
        this.communicatorLogger.set(communicatorLogger);
    }

    @Override
    public void disableLogging() {
        this.communicatorLogger.set(null);
    }

    // logging code

    private void logRequest(HttpRequest request, String requestId) {
        CommunicatorLogger logger = communicatorLogger.get();
        if (logger != null) {
            logRequest(request, requestId, logger);
        }
    }

    private void logRequest(HttpRequest request, String requestId, CommunicatorLogger logger) {
        try {
            String method = request.method();
            String uri = request.uri().toString();

            RequestLogMessageBuilder logMessageBuilder = new RequestLogMessageBuilder(requestId, method, uri,
                    bodyObfuscator.get(), headerObfuscator.get());

            HttpHeaders requestHeaders = request.headers();
            addHeaders(logMessageBuilder, requestHeaders);

            request.bodyPublisher().ifPresent(bodyPublisher -> {
                String contentType = getContentType(requestHeaders);

                // There are only two options: JSON or multipart; treat the latter as binary
                if (bodyPublisher instanceof JsonRequestBodyPublisher jsonBodyPublisher) {
                    logMessageBuilder.setBody(jsonBodyPublisher.json, contentType);
                } else {
                    logMessageBuilder.setBinaryContentBody(contentType);
                }
            });

            logger.log(logMessageBuilder.getMessage());

        } catch (Exception e) {
            logger.log(String.format("An error occurred trying to log request '%s'", requestId), e);
        }
    }

    private static void addHeaders(LogMessageBuilder logMessageBuilder, HttpHeaders headers) {
        if (headers != null) {
            Map<String, List<String>> headerMap = headers.map();
            for (Map.Entry<String, List<String>> entry : headerMap.entrySet()) {
                String headerName = entry.getKey();
                for (String headerValue : entry.getValue()) {
                    logMessageBuilder.addHeader(headerName, headerValue);
                }
            }
        }
    }

    private static String getContentType(HttpHeaders headers) {
        return headers.firstValue("Content-Type").orElse(null);
    }

    private static void setBody(LogMessageBuilder logMessageBuilder, String body, String contentType, boolean isBinaryContent) {
        if (body == null) {
            logMessageBuilder.setBody("", contentType);
        } else if (isBinaryContent) {
            logMessageBuilder.setBinaryContentBody(contentType);
        } else {
            logMessageBuilder.setBody(body, contentType);
        }
    }

    private static boolean isBinaryContent(String contentType) {
        return contentType != null
                && !contentType.startsWith("text/")
                && !contentType.contains("json")
                && !contentType.contains("xml");
    }

    private static void logError(String requestId, Exception error, long startTime, CommunicatorLogger logger) {
        if (logger != null) {
            final String messageTemplate = "Error occurred for outgoing request (requestId='%s', %d ms)";

            long endTime = System.currentTimeMillis();
            long duration = endTime - startTime;

            String message = String.format(messageTemplate, requestId, duration);

            logger.log(message, error);
        }
    }

    /**
     * A builder for {@link HttpClientConnection} objects.
     *
     * @author Rob Spoor
     */
    public static final class Builder {

        private final int connectTimeout;
        private final int socketTimeout;
        private ProxyConfiguration proxyConfiguration;
        private SSLContext sslContext;
        private SSLParameters sslParameters;

        /**
         * Creates a new builder.
         *
         * @param connectTimeout The connect timeout in milliseconds.
         * @param socketTimeout The socket timeout in milliseconds.
         */
        public Builder(int connectTimeout, int socketTimeout) {
            this.connectTimeout = connectTimeout;
            this.socketTimeout = socketTimeout;
        }

        /**
         * Sets the proxy configuration to use. Defaults to no proxy configuration.
         *
         * @param proxyConfiguration The proxy configuration to use.
         * @return This object.
         */
        public Builder withProxyConfiguration(ProxyConfiguration proxyConfiguration) {
            this.proxyConfiguration = proxyConfiguration;
            return this;
        }

        /**
         * Sets the SSL context to use.
         *
         * @param sslContext The SSL context to use.
         * @return This object.
         */
        public Builder withSslContext(SSLContext sslContext) {
            this.sslContext = sslContext;
            return this;
        }

        /**
         * Sets the HTTPS protocols to support. Defaults to {@link CommunicatorConfiguration#DEFAULT_HTTPS_PROTOCOLS}.
         * <p>
         * This method is mutually exclusive with {@link #withSslParameters(SSLParameters)}.
         *
         * @param httpsProtocols The HTTPS protocols to support.
         * @return This object.
         */
        public Builder withHttpsProtocols(Set<String> httpsProtocols) {
            return withSslParameters(createSslParameters(httpsProtocols));
        }

        /**
         * Sets the SSL parameters to use.
         * <p>
         * This method is mutually exclusive with {@link #withHttpsProtocols(Set)}.
         *
         * @param sslParameters The SSL parameters to use.
         * @return This object.
         */
        public Builder withSslParameters(SSLParameters sslParameters) {
            this.sslParameters = sslParameters;
            return this;
        }

        private static SSLParameters createSslParameters(Set<String> httpsProtocols) {
            Set<String> supportedProtocols = httpsProtocols != null && !httpsProtocols.isEmpty()
                    ? httpsProtocols
                    : CommunicatorConfiguration.DEFAULT_HTTPS_PROTOCOLS;

            return new SSLParameters(null, supportedProtocols.toArray(String[]::new));
        }

        /**
         * Creates a fully initialized {@link HttpClientConnection} object.
         *
         * @return The created {@link HttpClientConnection} object.
         */
        public HttpClientConnection build() {
            return new HttpClientConnection(this);
        }
    }

    private final class LoggingResponseBodySubscriber<T> implements HttpResponse.BodySubscriber<T> {

        private final HttpResponse.BodySubscriber<T> delegate;

        private final String requestId;
        private final long startTime;
        private final HttpResponse.ResponseInfo responseInfo;

        private final CommunicatorLogger logger;

        private final boolean captureBody;

        private ByteArrayOutputStream outputStream;
        private WritableByteChannel byteChannel;

        private LoggingResponseBodySubscriber(HttpResponse.BodySubscriber<T> delegate, String requestId, long startTime,
                                              HttpResponse.ResponseInfo responseInfo, CommunicatorLogger logger) {
            this.delegate = delegate;
            this.requestId = requestId;
            this.startTime = startTime;
            this.responseInfo = responseInfo;
            this.logger = logger;

            String contentType = getContentType(responseInfo.headers());
            this.captureBody = !isBinaryContent(contentType);
        }

        @Override
        public void onSubscribe(Subscription subscription) {
            if (captureBody) {
                outputStream = new ByteArrayOutputStream();
                byteChannel = Channels.newChannel(outputStream);
            }

            subscription.request(Long.MAX_VALUE);

            delegate.onSubscribe(subscription);
        }

        @Override
        public void onNext(List<ByteBuffer> item) {
            if (captureBody) {
                item.forEach(this::onNext);
            }

            delegate.onNext(item);
        }

        private void onNext(ByteBuffer item) {
            int position = item.position();
            int limit = item.limit();
            try {
                byteChannel.write(item);
            } catch (IOException e) {
                throw new UncheckedIOException(e);
            } finally {
                item.position(position).limit(limit);
            }
        }

        @Override
        public void onError(Throwable throwable) {
            log(throwable);

            delegate.onError(throwable);
        }

        @Override
        public void onComplete() {
            log(null);

            delegate.onComplete();
        }

        @Override
        public CompletionStage<T> getBody() {
            return delegate.getBody();
        }

        private void log(Throwable throwable) {
            String body = outputStream != null ? outputStream.toString(StandardCharsets.UTF_8) : null;
            logResponse(responseInfo, requestId, body, startTime, throwable, logger);
        }

        private void logResponse(HttpResponse.ResponseInfo responseInfo, String requestId, String body, long startTime, Throwable throwable,
                                 CommunicatorLogger logger) {

            long endTime = System.currentTimeMillis();
            long duration = endTime - startTime;

            try {
                int statusCode = responseInfo.statusCode();

                ResponseLogMessageBuilder logMessageBuilder = new ResponseLogMessageBuilder(requestId, statusCode, duration,
                        bodyObfuscator.get(), headerObfuscator.get());

                HttpHeaders responseHeaders = responseInfo.headers();
                addHeaders(logMessageBuilder, responseHeaders);

                String contentType = getContentType(responseHeaders);
                boolean isBinaryContent = isBinaryContent(contentType);

                setBody(logMessageBuilder, body, contentType, isBinaryContent);

                if (throwable == null) {
                    logger.log(logMessageBuilder.getMessage());
                } else {
                    logger.log(logMessageBuilder.getMessage(), throwable);
                }
            } catch (Exception e) {
                logger.log(String.format("An error occurred trying to log response '%s'", requestId), e);
            }
        }
    }

    private static final class JsonRequestBodyPublisher implements HttpRequest.BodyPublisher {

        private final HttpRequest.BodyPublisher delegate;
        private final String json;

        JsonRequestBodyPublisher(String json, Charset charset) {
            this.delegate = HttpRequest.BodyPublishers.ofString(json, charset);
            this.json = json;
        }

        @Override
        public void subscribe(Subscriber<? super ByteBuffer> subscriber) {
            delegate.subscribe(subscriber);
        }

        @Override
        public long contentLength() {
            return delegate.contentLength();
        }
    }

    private static final class MultipartBodyPublisher implements HttpRequest.BodyPublisher {

        private static final String LINE_BREAK = "\r\n";

        private final HttpRequest.BodyPublisher delegate;

        private MultipartBodyPublisher(MultipartFormDataObject multipart) {
            String boundary = multipart.getBoundary();

            List<HttpRequest.BodyPublisher> parts = new ArrayList<>();
            for (Map.Entry<String, String> entry : multipart.getValues().entrySet()) {
                addPart(entry.getKey(), entry.getValue(), boundary, parts);
            }
            for (Map.Entry<String, UploadableFile> entry : multipart.getFiles().entrySet()) {
                addPart(entry.getKey(), entry.getValue(), boundary, parts);
            }
            parts.add(HttpRequest.BodyPublishers.ofString("--" + boundary + "--" + LINE_BREAK, CHARSET));

            this.delegate = HttpRequest.BodyPublishers.concat(parts.toArray(HttpRequest.BodyPublisher[]::new));
        }

        private void addPart(String name, String value, String boundary, List<HttpRequest.BodyPublisher> parts) {
            StringBuilder part = new StringBuilder();
            addBoundary(boundary, part);
            addContentDisposition(name, null, part);
            part.append(LINE_BREAK);
            part.append(value).append(LINE_BREAK);
            parts.add(HttpRequest.BodyPublishers.ofString(part.toString(), CHARSET));
        }

        private void addPart(String name, UploadableFile file, String boundary, List<HttpRequest.BodyPublisher> parts) {
            StringBuilder part = new StringBuilder();
            addBoundary(boundary, part);
            addContentDisposition(name, file.getFileName(), part);
            addContentType(file.getContentType(), part);
            part.append(LINE_BREAK);
            parts.add(HttpRequest.BodyPublishers.ofString(part.toString(), CHARSET));
            parts.add(new UploadableFileContentBodyPublisher(file));
            parts.add(HttpRequest.BodyPublishers.ofString(LINE_BREAK, CHARSET));
        }

        private void addBoundary(String boundary, StringBuilder part) {
            part.append("--").append(boundary).append(LINE_BREAK);
        }

        private void addContentDisposition(String name, String fileName, StringBuilder part) {
            part.append("Content-Disposition: form-data; name=\"").append(name).append('"');
            if (fileName != null) {
                part.append("; filename=\"").append(fileName).append('"');
            }
            part.append(LINE_BREAK);
        }

        private void addContentType(String contentType, StringBuilder part) {
            part.append("Content-Type: ").append(contentType).append(LINE_BREAK);
        }

        @Override
        public void subscribe(Subscriber<? super ByteBuffer> subscriber) {
            delegate.subscribe(subscriber);
        }

        @Override
        public long contentLength() {
            return delegate.contentLength();
        }
    }

    private static final class UploadableFileContentBodyPublisher implements HttpRequest.BodyPublisher {

        private final HttpRequest.BodyPublisher delegate;
        private final long contentLength;

        private UploadableFileContentBodyPublisher(UploadableFile file) {
            this.delegate = HttpRequest.BodyPublishers.ofInputStream(file::getContent);
            this.contentLength = file.getContentLength();
        }

        @Override
        public void subscribe(Subscriber<? super ByteBuffer> subscriber) {
            delegate.subscribe(subscriber);
        }

        @Override
        public long contentLength() {
            return contentLength;
        }
    }
 }
No results found