connectors/jdk-connector/src/main/java/org/glassfish/jersey/jdk/connector/internal/JdkConnector.java

/*
 * Copyright (c) 2015, 2019 Oracle and/or its affiliates. All rights reserved.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License v. 2.0, which is available at
 * http://www.eclipse.org/legal/epl-2.0.
 *
 * This Source Code may also be made available under the following Secondary
 * Licenses when the conditions for such availability set forth in the
 * Eclipse Public License v. 2.0 are satisfied: GNU General Public License,
 * version 2 with the GNU Classpath Exception, which is available at
 * https://www.gnu.org/software/classpath/license.html.
 *
 * SPDX-License-Identifier: EPL-2.0 OR GPL-2.0 WITH Classpath-exception-2.0
 */

package org.glassfish.jersey.jdk.connector.internal;

import java.io.IOException;
import java.net.CookieManager;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import javax.ws.rs.ProcessingException;
import javax.ws.rs.client.Client;
import javax.ws.rs.core.Configuration;
import javax.ws.rs.core.Response;

import org.glassfish.jersey.client.ClientProperties;
import org.glassfish.jersey.client.ClientRequest;
import org.glassfish.jersey.client.ClientResponse;
import org.glassfish.jersey.client.RequestEntityProcessing;
import org.glassfish.jersey.client.spi.AsyncConnectorCallback;
import org.glassfish.jersey.client.spi.Connector;

/**
 * @author Petr Janouch
 */
public class JdkConnector implements Connector {

    private final HttpConnectionPool httpConnectionPool;
    private final ConnectorConfiguration connectorConfiguration;

    public JdkConnector(Client client, Configuration config) {
        connectorConfiguration = new ConnectorConfiguration(client, config);
        CookieManager cookieManager = new CookieManager();
        cookieManager.setCookiePolicy(connectorConfiguration.getCookiePolicy());
        httpConnectionPool = new HttpConnectionPool(connectorConfiguration, cookieManager);
    }

    @Override
    public ClientResponse apply(ClientRequest request) {

        Future<?> future = apply(request, new AsyncConnectorCallback() {
            @Override
            public void response(ClientResponse response) {

            }

            @Override
            public void failure(Throwable failure) {

            }
        });

        try {
            return (ClientResponse) future.get();
        } catch (Exception e) {
            throw new ProcessingException(unwrapExecutionException(e));
        }
    }

    private Throwable unwrapExecutionException(Throwable failure) {
        return (failure != null && failure instanceof ExecutionException) ? failure.getCause() : failure;
    }

    @Override
    public Future<?> apply(final ClientRequest request, final AsyncConnectorCallback callback) {
        final CompletableFuture<ClientResponse> responseFuture = new CompletableFuture<>();
        // just so we don't have to drag around both the future and callback
        final AsyncConnectorCallback internalCallback = new AsyncConnectorCallback() {
            @Override
            public void response(ClientResponse response) {
                callback.response(response);
                responseFuture.complete(response);
            }

            @Override
            public void failure(Throwable failure) {
                Throwable actualFailure = unwrapExecutionException(failure);
                callback.failure(actualFailure);
                responseFuture.completeExceptionally(actualFailure);
            }
        };

        final HttpRequest httpRequest = createHttpRequest(request);

        if (httpRequest.getBodyMode() == HttpRequest.BodyMode.BUFFERED) {
            writeBufferedEntity(request, httpRequest, internalCallback);
        }

        if (httpRequest.getBodyMode() == HttpRequest.BodyMode.BUFFERED
                || httpRequest.getBodyMode() == HttpRequest.BodyMode.NONE) {
            send(request, httpRequest, internalCallback);
        }

        if (httpRequest.getBodyMode() == HttpRequest.BodyMode.CHUNKED) {

            /* We wait with sending the request header until the body stream has been touched.
             This is because of javax.ws.rs.ext.MessageBodyWriter, which says:

             "The message header map is mutable but any changes must be made before writing to the output stream since
              the headers will be flushed prior to writing the message body"

              This means that the headers can change until body output stream is used.
              */
            final InterceptingOutputStream bodyStream = new InterceptingOutputStream(httpRequest.getBodyStream(),
                    // send the prepared request when the stream is touched for the first time
                    () -> send(request, httpRequest, internalCallback));

            request.setStreamProvider(contentLength -> bodyStream);
            try {
                request.writeEntity();
            } catch (IOException e) {
                internalCallback.failure(e);
            }
        }

        return responseFuture;
    }

    private void writeBufferedEntity(ClientRequest request, final HttpRequest httpRequest, AsyncConnectorCallback callback) {
        request.setStreamProvider(contentLength -> httpRequest.getBodyStream());
        try {
            request.writeEntity();
        } catch (IOException e) {
            callback.failure(e);
        }
    }

    private void send(final ClientRequest request, final HttpRequest httpRequest, final AsyncConnectorCallback callback) {
        translateHeaders(request, httpRequest);
        final RedirectHandler redirectHandler = new RedirectHandler(httpConnectionPool, httpRequest, connectorConfiguration);
        httpConnectionPool.send(httpRequest, new CompletionHandler<HttpResponse>() {

            @Override
            public void failed(Throwable throwable) {
                callback.failure(throwable);
            }

            @Override
            public void completed(HttpResponse result) {
                redirectHandler.handleRedirects(result, new CompletionHandler<HttpResponse>() {
                    @Override
                    public void failed(Throwable throwable) {
                        Throwable actualFailure = unwrapExecutionException(throwable);
                        callback.failure(actualFailure);
                    }

                    @Override
                    public void completed(HttpResponse result) {
                        ClientResponse response = translateResponse(request, result, redirectHandler.getLastRequestUri());
                        callback.response(response);
                    }
                });
            }
        });
    }

    private HttpRequest createHttpRequest(ClientRequest request) {
        Object entity = request.getEntity();

        if (entity == null) {
            return HttpRequest.createBodyless(request.getMethod(), request.getUri());
        }

        RequestEntityProcessing entityProcessing = request.resolveProperty(
                ClientProperties.REQUEST_ENTITY_PROCESSING, RequestEntityProcessing.class);

        HttpRequest httpRequest;
        if (entityProcessing != null && entityProcessing == RequestEntityProcessing.CHUNKED) {
            httpRequest = HttpRequest.createChunked(request.getMethod(), request.getUri(), connectorConfiguration.getChunkSize());
        } else {
            httpRequest = HttpRequest.createBuffered(request.getMethod(), request.getUri());
        }

        return httpRequest;
    }

    private Map<String, List<String>> translateHeaders(ClientRequest clientRequest, HttpRequest httpRequest) {
        Map<String, List<String>> headers = httpRequest.getHeaders();
        for (Map.Entry<String, List<String>> header : clientRequest.getStringHeaders().entrySet()) {
            List<String> values = new ArrayList<>(header.getValue());
            headers.put(header.getKey(), values);
        }

        return headers;
    }

    private ClientResponse translateResponse(final ClientRequest requestContext,
                                             final HttpResponse httpResponse,
                                             URI requestUri) {

        Response.StatusType statusType = new Response.StatusType() {
            @Override
            public int getStatusCode() {
                return httpResponse.getStatusCode();
            }

            @Override
            public Response.Status.Family getFamily() {
                return Response.Status.Family.familyOf(httpResponse.getStatusCode());
            }

            @Override
            public String getReasonPhrase() {
                return httpResponse.getReasonPhrase();
            }
        };

        ClientResponse responseContext = new ClientResponse(statusType, requestContext, requestUri);

        Map<String, List<String>> headers = httpResponse.getHeaders();
        for (Map.Entry<String, List<String>> entry : headers.entrySet()) {
            for (String value : entry.getValue()) {
                responseContext.getHeaders().add(entry.getKey(), value);
            }
        }

        responseContext.setEntityStream(httpResponse.getBodyStream());
        return responseContext;
    }

    @Override
    public String getName() {
        return "JDK connector";
    }

    @Override
    public void close() {
        httpConnectionPool.close();
    }
}