/*

 * Copyright (c) 2015, 2018, Oracle and/or its affiliates. All rights reserved.

 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

 *

 * This code is free software; you can redistribute it and/or modify it

 * under the terms of the GNU General Public License version 2 only, as

 * published by the Free Software Foundation.  Oracle designates this

 * particular file as subject to the "Classpath" exception as provided

 * by Oracle in the LICENSE file that accompanied this code.

 *

 * This code is distributed in the hope that it will be useful, but WITHOUT

 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

 * version 2 for more details (a copy is included in the LICENSE file that

 * accompanied this code).

 *

 * You should have received a copy of the GNU General Public License version

 * 2 along with this work; if not, write to the Free Software Foundation,

 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

 *

 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

 * or visit www.oracle.com if you need additional information or have any

 * questions.

 */

package jdk.internal.net.http;

import javax.net.ssl.SSLContext;

import javax.net.ssl.SSLParameters;

import java.io.IOException;

import java.io.UncheckedIOException;

import java.lang.ref.Reference;

import java.lang.ref.WeakReference;

import java.net.Authenticator;

import java.net.ConnectException;

import java.net.CookieHandler;

import java.net.ProxySelector;

import java.net.http.HttpConnectTimeoutException;

import java.net.http.HttpTimeoutException;

import java.nio.ByteBuffer;

import java.nio.channels.CancelledKeyException;

import java.nio.channels.ClosedChannelException;

import java.nio.channels.SelectableChannel;

import java.nio.channels.SelectionKey;

import java.nio.channels.Selector;

import java.nio.channels.SocketChannel;

import java.security.AccessControlContext;

import java.security.AccessController;

import java.security.NoSuchAlgorithmException;

import java.security.PrivilegedAction;

import java.time.Duration;

import java.time.Instant;

import java.time.temporal.ChronoUnit;

import java.util.ArrayList;

import java.util.HashSet;

import java.util.Iterator;

import java.util.LinkedList;

import java.util.List;

import java.util.Objects;

import java.util.Optional;

import java.util.Set;

import java.util.TreeSet;

import java.util.concurrent.CompletableFuture;

import java.util.concurrent.CompletionException;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.Executor;

import java.util.concurrent.Executors;

import java.util.concurrent.ThreadFactory;

import java.util.concurrent.atomic.AtomicInteger;

import java.util.concurrent.atomic.AtomicLong;

import java.util.function.BooleanSupplier;

import java.util.stream.Stream;

import java.net.http.HttpClient;

import java.net.http.HttpRequest;

import java.net.http.HttpResponse;

import java.net.http.HttpResponse.BodyHandler;

import java.net.http.HttpResponse.PushPromiseHandler;

import java.net.http.WebSocket;

import jdk.internal.net.http.common.BufferSupplier;

import jdk.internal.net.http.common.Log;

import jdk.internal.net.http.common.Logger;

import jdk.internal.net.http.common.Pair;

import jdk.internal.net.http.common.Utils;

import jdk.internal.net.http.common.OperationTrackers.Trackable;

import jdk.internal.net.http.common.OperationTrackers.Tracker;

import jdk.internal.net.http.websocket.BuilderImpl;

import jdk.internal.misc.InnocuousThread;

/**

 * Client implementation. Contains all configuration information and also

 * the selector manager thread which allows async events to be registered

 * and delivered when they occur. See AsyncEvent.

 */

final class HttpClientImpl extends HttpClient implements Trackable {

    static final boolean DEBUGELAPSED = Utils.TESTING || Utils.DEBUG;  // dev flag

    static final boolean DEBUGTIMEOUT = false; // dev flag

    final Logger debug = Utils.getDebugLogger(this::dbgString, Utils.DEBUG);

    final Logger debugelapsed = Utils.getDebugLogger(this::dbgString, DEBUGELAPSED);

    final Logger debugtimeout = Utils.getDebugLogger(this::dbgString, DEBUGTIMEOUT);

    static final AtomicLong CLIENT_IDS = new AtomicLong();

    // Define the default factory as a static inner class

    // that embeds all the necessary logic to avoid

    // the risk of using a lambda that might keep a reference on the

    // HttpClient instance from which it was created (helps with

    // heapdump analysis).

    private static final class DefaultThreadFactory implements ThreadFactory {

        private final String namePrefix;

        private final AtomicInteger nextId = new AtomicInteger();

        DefaultThreadFactory(long clientID) {

            namePrefix = "HttpClient-" + clientID + "-Worker-";

        }

        @Override

        public Thread newThread(Runnable r) {

            String name = namePrefix + nextId.getAndIncrement();

            Thread t;

            if (System.getSecurityManager() == null) {

                t = new Thread(null, r, name, 0, false);

            } else {

                t = InnocuousThread.newThread(name, r);

            }

            t.setDaemon(true);

            return t;

        }

    }

    /**

     * A DelegatingExecutor is an executor that delegates tasks to

     * a wrapped executor when it detects that the current thread

     * is the SelectorManager thread. If the current thread is not

     * the selector manager thread the given task is executed inline.

     */

    final static class DelegatingExecutor implements Executor {

        private final BooleanSupplier isInSelectorThread;

        private final Executor delegate;

        DelegatingExecutor(BooleanSupplier isInSelectorThread, Executor delegate) {

            this.isInSelectorThread = isInSelectorThread;

            this.delegate = delegate;

        }

        Executor delegate() {

            return delegate;

        }

        @Override

        public void execute(Runnable command) {

            if (isInSelectorThread.getAsBoolean()) {

                delegate.execute(command);

            } else {

                command.run();

            }

        }

    }

    private final CookieHandler cookieHandler;

    private final Duration connectTimeout;

    private final Redirect followRedirects;

    private final Optional<ProxySelector> userProxySelector;

    private final ProxySelector proxySelector;

    private final Authenticator authenticator;

    private final Version version;

    private final ConnectionPool connections;

    private final DelegatingExecutor delegatingExecutor;

    private final boolean isDefaultExecutor;

    // Security parameters

    private final SSLContext sslContext;

    private final SSLParameters sslParams;

    private final SelectorManager selmgr;

    private final FilterFactory filters;

    private final Http2ClientImpl client2;

    private final long id;

    private final String dbgTag;

    // The SSL DirectBuffer Supplier provides the ability to recycle

    // buffers used between the socket reader and the SSLEngine, or

    // more precisely between the SocketTube publisher and the

    // SSLFlowDelegate reader.

    private final SSLDirectBufferSupplier sslBufferSupplier

            = new SSLDirectBufferSupplier(this);

    // This reference is used to keep track of the facade HttpClient

    // that was returned to the application code.

    // It makes it possible to know when the application no longer

    // holds any reference to the HttpClient.

    // Unfortunately, this information is not enough to know when

    // to exit the SelectorManager thread. Because of the asynchronous

    // nature of the API, we also need to wait until all pending operations

    // have completed.

    private final WeakReference<HttpClientFacade> facadeRef;

    // This counter keeps track of the number of operations pending

    // on the HttpClient. The SelectorManager thread will wait

    // until there are no longer any pending operations and the

    // facadeRef is cleared before exiting.

    //

    // The pendingOperationCount is incremented every time a send/sendAsync

    // operation is invoked on the HttpClient, and is decremented when

    // the HttpResponse<T> object is returned to the user.

    // However, at this point, the body may not have been fully read yet.

    // This is the case when the response T is implemented as a streaming

    // subscriber (such as an InputStream).

    //

    // To take care of this issue the pendingOperationCount will additionally

    // be incremented/decremented in the following cases:

    //

    // 1. For HTTP/2  it is incremented when a stream is added to the

    //    Http2Connection streams map, and decreased when the stream is removed

    //    from the map. This should also take care of push promises.

    // 2. For WebSocket the count is increased when creating a

    //    DetachedConnectionChannel for the socket, and decreased

    //    In addition, the HttpClient facade is passed to the WebSocket builder,

    //    (instead of the client implementation delegate).

    // 3. For HTTP/1.1 the count is incremented before starting to parse the body

    //    response, and decremented when the parser has reached the end of the

    //    response body flow.

    //

    // This should ensure that the selector manager thread remains alive until

    // the response has been fully received or the web socket is closed.

    private final AtomicLong pendingOperationCount = new AtomicLong();

    private final AtomicLong pendingWebSocketCount = new AtomicLong();

    private final AtomicLong pendingHttpRequestCount = new AtomicLong();

    private final AtomicLong pendingHttp2StreamCount = new AtomicLong();

    /** A Set of, deadline first, ordered timeout events. */

    private final TreeSet<TimeoutEvent> timeouts;

    /**

     * This is a bit tricky:

     * 1. an HttpClientFacade has a final HttpClientImpl field.

     * 2. an HttpClientImpl has a final WeakReference<HttpClientFacade> field,

     *    where the referent is the facade created for that instance.

     * 3. We cannot just create the HttpClientFacade in the HttpClientImpl

     *    constructor, because it would be only weakly referenced and could

     *    be GC'ed before we can return it.

     * The solution is to use an instance of SingleFacadeFactory which will

     * allow the caller of new HttpClientImpl(...) to retrieve the facade

     * after the HttpClientImpl has been created.

     */

    private static final class SingleFacadeFactory {

        HttpClientFacade facade;

        HttpClientFacade createFacade(HttpClientImpl impl) {

            assert facade == null;

            return (facade = new HttpClientFacade(impl));

        }

    }

    static HttpClientFacade create(HttpClientBuilderImpl builder) {

        SingleFacadeFactory facadeFactory = new SingleFacadeFactory();

        HttpClientImpl impl = new HttpClientImpl(builder, facadeFactory);

        impl.start();

        assert facadeFactory.facade != null;

        assert impl.facadeRef.get() == facadeFactory.facade;

        return facadeFactory.facade;

    }

    private HttpClientImpl(HttpClientBuilderImpl builder,

                           SingleFacadeFactory facadeFactory) {

        id = CLIENT_IDS.incrementAndGet();

        dbgTag = "HttpClientImpl(" + id +")";

        if (builder.sslContext == null) {

            try {

                sslContext = SSLContext.getDefault();

            } catch (NoSuchAlgorithmException ex) {

                throw new InternalError(ex);

            }

        } else {

            sslContext = builder.sslContext;

        }

        Executor ex = builder.executor;

        if (ex == null) {

            ex = Executors.newCachedThreadPool(new DefaultThreadFactory(id));

            isDefaultExecutor = true;

        } else {

            isDefaultExecutor = false;

        }

        delegatingExecutor = new DelegatingExecutor(this::isSelectorThread, ex);

        facadeRef = new WeakReference<>(facadeFactory.createFacade(this));

        client2 = new Http2ClientImpl(this);

        cookieHandler = builder.cookieHandler;

        connectTimeout = builder.connectTimeout;

        followRedirects = builder.followRedirects == null ?

                Redirect.NEVER : builder.followRedirects;

        this.userProxySelector = Optional.ofNullable(builder.proxy);

        this.proxySelector = userProxySelector

                .orElseGet(HttpClientImpl::getDefaultProxySelector);

        if (debug.on())

            debug.log("proxySelector is %s (user-supplied=%s)",

                      this.proxySelector, userProxySelector.isPresent());

        authenticator = builder.authenticator;

        if (builder.version == null) {

            version = HttpClient.Version.HTTP_2;

        } else {

            version = builder.version;

        }

        if (builder.sslParams == null) {

            sslParams = getDefaultParams(sslContext);

        } else {

            sslParams = builder.sslParams;

        }

        connections = new ConnectionPool(id);

        connections.start();

        timeouts = new TreeSet<>();

        try {

            selmgr = new SelectorManager(this);

        } catch (IOException e) {

            // unlikely

            throw new InternalError(e);

        }

        selmgr.setDaemon(true);

        filters = new FilterFactory();

        initFilters();

        assert facadeRef.get() != null;

    }

    private void start() {

        selmgr.start();

    }

    // Called from the SelectorManager thread, just before exiting.

    // Clears the HTTP/1.1 and HTTP/2 cache, ensuring that the connections

    // that may be still lingering there are properly closed (and their

    // possibly still opened SocketChannel released).

    private void stop() {

        // Clears HTTP/1.1 cache and close its connections

        connections.stop();

        // Clears HTTP/2 cache and close its connections.

        client2.stop();

    }

    private static SSLParameters getDefaultParams(SSLContext ctx) {

        SSLParameters params = ctx.getSupportedSSLParameters();

        String[] protocols = params.getProtocols();

        boolean found13 = false;

        for (String proto : protocols) {

            if (proto.equals("TLSv1.3")) {

                found13 = true;

                break;

            }

        }

        if (found13)

            params.setProtocols(new String[] {"TLSv1.3", "TLSv1.2"});

        else

            params.setProtocols(new String[] {"TLSv1.2"});

        return params;

    }

    private static ProxySelector getDefaultProxySelector() {

        PrivilegedAction<ProxySelector> action = ProxySelector::getDefault;

        return AccessController.doPrivileged(action);

    }

    // Returns the facade that was returned to the application code.

    // May be null if that facade is no longer referenced.

    final HttpClientFacade facade() {

        return facadeRef.get();

    }

    // Increments the pendingOperationCount.

    final long reference() {

        pendingHttpRequestCount.incrementAndGet();

        return pendingOperationCount.incrementAndGet();

    }

    // Decrements the pendingOperationCount.

    final long unreference() {

        final long count = pendingOperationCount.decrementAndGet();

        final long httpCount = pendingHttpRequestCount.decrementAndGet();

        final long http2Count = pendingHttp2StreamCount.get();

        final long webSocketCount = pendingWebSocketCount.get();

        if (count == 0 && facade() == null) {

            selmgr.wakeupSelector();

        }

        assert httpCount >= 0 : "count of HTTP/1.1 operations < 0";

        assert http2Count >= 0 : "count of HTTP/2 operations < 0";

        assert webSocketCount >= 0 : "count of WS operations < 0";

        assert count >= 0 : "count of pending operations < 0";

        return count;

    }

    // Increments the pendingOperationCount.

    final long streamReference() {

        pendingHttp2StreamCount.incrementAndGet();

        return pendingOperationCount.incrementAndGet();

    }

    // Decrements the pendingOperationCount.

    final long streamUnreference() {

        final long count = pendingOperationCount.decrementAndGet();

        final long http2Count = pendingHttp2StreamCount.decrementAndGet();

        final long httpCount = pendingHttpRequestCount.get();

        final long webSocketCount = pendingWebSocketCount.get();

        if (count == 0 && facade() == null) {

            selmgr.wakeupSelector();

        }

        assert httpCount >= 0 : "count of HTTP/1.1 operations < 0";

        assert http2Count >= 0 : "count of HTTP/2 operations < 0";

        assert webSocketCount >= 0 : "count of WS operations < 0";

        assert count >= 0 : "count of pending operations < 0";

        return count;

    }

    // Increments the pendingOperationCount.

    final long webSocketOpen() {

        pendingWebSocketCount.incrementAndGet();

        return pendingOperationCount.incrementAndGet();

    }

    // Decrements the pendingOperationCount.

    final long webSocketClose() {

        final long count = pendingOperationCount.decrementAndGet();

        final long webSocketCount = pendingWebSocketCount.decrementAndGet();

        final long httpCount = pendingHttpRequestCount.get();

        final long http2Count = pendingHttp2StreamCount.get();

        if (count == 0 && facade() == null) {

            selmgr.wakeupSelector();

        }

        assert httpCount >= 0 : "count of HTTP/1.1 operations < 0";

        assert http2Count >= 0 : "count of HTTP/2 operations < 0";

        assert webSocketCount >= 0 : "count of WS operations < 0";

        assert count >= 0 : "count of pending operations < 0";

        return count;

    }

    // Returns the pendingOperationCount.

    final long referenceCount() {

        return pendingOperationCount.get();

    }

    final static class HttpClientTracker implements Tracker {

        final AtomicLong httpCount;

        final AtomicLong http2Count;

        final AtomicLong websocketCount;

        final AtomicLong operationsCount;

        final Reference<?> reference;

        final String name;

        HttpClientTracker(AtomicLong http,

                          AtomicLong http2,

                          AtomicLong ws,

                          AtomicLong ops,

                          Reference<?> ref,

                          String name) {

            this.httpCount = http;

            this.http2Count = http2;

            this.websocketCount = ws;

            this.operationsCount = ops;

            this.reference = ref;

            this.name = name;

        }

        @Override

        public long getOutstandingOperations() {

            return operationsCount.get();

        }

        @Override

        public long getOutstandingHttpOperations() {

            return httpCount.get();

        }

        @Override

        public long getOutstandingHttp2Streams() { return http2Count.get(); }

        @Override

        public long getOutstandingWebSocketOperations() {

            return websocketCount.get();

        }

        @Override

        public boolean isFacadeReferenced() {

            return reference.get() != null;

        }

        @Override

        public String getName() {

            return name;

        }

    }

    public Tracker getOperationsTracker() {

        return new HttpClientTracker(pendingHttpRequestCount,

                pendingHttp2StreamCount,

                pendingWebSocketCount,

                pendingOperationCount,

                facadeRef,

                dbgTag);

    }

    // Called by the SelectorManager thread to figure out whether it's time

    // to terminate.

    final boolean isReferenced() {

        HttpClient facade = facade();

        return facade != null || referenceCount() > 0;

    }

    /**

     * Wait for activity on given exchange.

     * The following occurs in the SelectorManager thread.

     *

     *  1) add to selector