// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package ai.vespa.feed.client;

import okhttp3.Call;
import okhttp3.Callback;
import okhttp3.ConnectionPool;
import okhttp3.ConnectionSpec;
import okhttp3.MediaType;
import okhttp3.OkHttp;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.RequestBody;
import okhttp3.Response;
import okhttp3.internal.concurrent.TaskRunner;
import okhttp3.internal.connection.RealConnectionPool;
import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.client.HttpClientTransport;
import org.eclipse.jetty.client.api.Result;
import org.eclipse.jetty.client.util.BufferingResponseListener;
import org.eclipse.jetty.client.util.BytesContentProvider;
import org.eclipse.jetty.http.HttpField;
import org.eclipse.jetty.http2.client.HTTP2Client;
import org.eclipse.jetty.http2.client.http.HttpClientTransportOverHTTP2;
import org.eclipse.jetty.util.ssl.SslContextFactory;
import org.jetbrains.annotations.NotNull;
import sun.security.ssl.SSLSocketFactoryImpl;

import javax.net.ssl.SSLEngine;
import javax.net.ssl.TrustManager;
import javax.net.ssl.TrustManagerFactory;
import javax.net.ssl.X509ExtendedTrustManager;
import java.io.IOException;
import java.net.Socket;
import java.net.URI;
import java.security.GeneralSecurityException;
import java.security.cert.CertificateException;
import java.security.cert.X509Certificate;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author jonmv
 */
public class OkCluster implements Cluster {

    private final List<Endpoint> endpoints = new ArrayList<>();

    OkCluster(FeedClientBuilder builder) {
        for (URI endpoint : builder.endpoints)
            for (int i = 0; i < builder.connectionsPerEndpoint; i++)
                endpoints.add(new Endpoint(createOkHttpClient(builder), endpoint));
    }

    private static OkHttpClient createOkHttpClient(FeedClientBuilder builder) {
        try {
            return new OkHttpClient.Builder().connectTimeout(15, TimeUnit.SECONDS)
                                             .callTimeout(5, TimeUnit.MINUTES)
                                             .readTimeout(30, TimeUnit.SECONDS)
                                             .writeTimeout(30, TimeUnit.SECONDS)
                                             .followRedirects(false)
                                             //.hostnameVerifier(builder.hostnameVerifier)
                                             .retryOnConnectionFailure(false)
                                             .sslSocketFactory(builder.constructSslContext().getSocketFactory(),
                                                               new X509ExtendedTrustManager() {
                                                                   @Override public void checkClientTrusted(X509Certificate[] chain, String authType, Socket socket) throws CertificateException { }
                                                                   @Override public void checkServerTrusted(X509Certificate[] chain, String authType, Socket socket) throws CertificateException { }
                                                                   @Override public void checkClientTrusted(X509Certificate[] chain, String authType, SSLEngine engine) throws CertificateException { }
                                                                   @Override public void checkServerTrusted(X509Certificate[] chain, String authType, SSLEngine engine) throws CertificateException { }
                                                                   @Override public void checkClientTrusted(X509Certificate[] chain, String authType) throws CertificateException { }
                                                                   @Override public void checkServerTrusted(X509Certificate[] chain, String authType) throws CertificateException { }
                                                                   @Override public X509Certificate[] getAcceptedIssuers() { return new X509Certificate[0]; } })
                    .build();

        }
        catch (Exception e) {
            throw new IllegalStateException(e);
        }
    }


    @Override
    public void dispatch(HttpRequest request, CompletableFuture<HttpResponse> vessel) {
        int index = 0;
        int min = Integer.MAX_VALUE;
        for (int i = 0; i < endpoints.size(); i++)
            if (endpoints.get(i).inflight.get() < min) {
                index = i;
                min = endpoints.get(i).inflight.get();
            }

        Endpoint endpoint = endpoints.get(index);
        endpoint.inflight.incrementAndGet();
        try {
            Request.Builder okRequest = new Request.Builder().method(request.method(),
                                                                     RequestBody.create(request.body(),
                                                                                        MediaType.parse("application/json")))
                                                             .url(endpoint.uri.resolve(request.path()).toString());
            request.headers().forEach((name, value) -> okRequest.header(name, value.get()));
            endpoint.client.newCall(okRequest.build()).enqueue(new Callback() {
                @Override public void onFailure(@NotNull Call call, @NotNull IOException e) {
                    vessel.completeExceptionally(e);
                }
                @Override public void onResponse(@NotNull Call call, @NotNull Response response) throws IOException {
                    vessel.complete(HttpResponse.of(response.code(), response.body().bytes()));
                }
            });
        }
        catch (Throwable thrown) {
            vessel.completeExceptionally(thrown);
        }
        vessel.whenComplete((__, ___) -> endpoint.inflight.decrementAndGet());
    }


    @Override
    public void close() {
        Throwable thrown = null;
        for (Endpoint endpoint : endpoints)
            try {
                //endpoint.client.
            }
            catch (Throwable t) {
                if (thrown == null) thrown = t;
                else thrown.addSuppressed(t);
            }
        if (thrown != null) throw new RuntimeException(thrown);
    }


    private static class Endpoint {

        private final OkHttpClient client;
        private final AtomicInteger inflight = new AtomicInteger(0);
        private final URI uri;

        private Endpoint(OkHttpClient client, URI uri) {
            this.client = client;
            this.uri = uri;
        }
    }

}

@@ -202,102 +202,17 @@
 tp3.
-ConnectionPool;%0Aimport okhttp3.ConnectionSpec;%0Aimport okhttp3.MediaType;%0Aimport okhttp3.OkHttp
+MediaType
 ;%0Aim
@@ -327,780 +327,73 @@
 rt o
-khttp3.internal.concurrent.TaskRunner;%0Aimport okhttp3.internal.connection.RealConnectionPool;%0Aimport org.eclipse.jetty.client.HttpClient;%0Aimport org.eclipse.jetty.client.HttpClientTransport;%0Aimport org.eclipse.jetty.client.api.Result;%0Aimport org.eclipse.jetty.client.util.BufferingResponseListener;%0Aimport org.eclipse.jetty.client.util.BytesContentProvider;%0Aimport org.eclipse.jetty.http.HttpField;%0Aimport org.eclipse.jetty.http2.client.HTTP2Client;%0Aimport org.eclipse.jetty.http2.client.http.HttpClientTransportOverHTTP2;%0Aimport org.eclipse.jetty.util.ssl.SslContextFactory;%0Aimport org.jetbrains.annotations.NotNull;%0Aimport sun.security.ssl.SSLSocketFactoryImpl;%0A%0Aimport javax.net.ssl.SSLEngine;%0Aimport javax.net.ssl.TrustManager;%0Aimport javax.net.ssl.TrustManagerFactory
+rg.jetbrains.annotations.NotNull;%0A%0Aimport javax.net.ssl.SSLEngine
 ;%0Aim
@@ -514,55 +514,8 @@
 RI;%0A
-import java.security.GeneralSecurityException;%0A
 impo
@@ -633,33 +633,8 @@
 st;%0A
-import java.util.Arrays;%0A
 impo

Clean up imports