/*

 * Copyright (c) 2015, 2017, 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.incubator.http;

import javax.net.ssl.SSLContext;

import javax.net.ssl.SSLParameters;

import java.io.IOException;

import java.lang.System.Logger.Level;

import java.lang.ref.WeakReference;

import java.net.Authenticator;

import java.net.CookieHandler;

import java.net.NetPermission;

import java.net.ProxySelector;

import java.net.URI;

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.time.Instant;

import java.time.temporal.ChronoUnit;

import java.util.ArrayList;

import java.util.HashSet;

import java.util.Iterator;

import java.util.List;

import java.util.Optional;

import java.util.Set;

import java.util.TreeSet;

import java.util.concurrent.CompletableFuture;

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.stream.Stream;

import jdk.incubator.http.HttpResponse.BodyHandler;

import jdk.incubator.http.HttpResponse.MultiSubscriber;

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

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

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

import jdk.incubator.http.internal.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.

 */

class HttpClientImpl extends HttpClient {

    static final boolean DEBUG = Utils.DEBUG;  // Revisit: temporary dev flag.

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

    static final boolean DEBUGTIMEOUT = false; // Revisit: temporary dev flag.

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

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

    final System.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;

        }

    }

    private final CookieHandler cookieHandler;

    private final Redirect followRedirects;

    private final ProxySelector proxySelector;

    private final Authenticator authenticator;

    private final Version version;

    private final ConnectionPool connections;

    private final Executor executor;

    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;

    // 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

    //    when the the channel is closed.

    //    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();

    /** 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 {

            ex = builder.executor;

            isDefaultExecutor = false;

        }

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

        client2 = new Http2ClientImpl(this);

        executor = ex;

        cookieHandler = builder.cookieHandler;

        followRedirects = builder.followRedirects == null ?

                Redirect.NEVER : builder.followRedirects;

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

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

        return params;

    }

    // 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 webSocketCount = pendingWebSocketCount.get();

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

            selmgr.wakeupSelector();

        }

        assert httpCount >= 0 : "count of HTTP 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();

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

            selmgr.wakeupSelector();

        }

        assert httpCount >= 0 : "count of HTTP 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();

    }

    // 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

     *  2) If selector fires for this exchange then

     *     call AsyncEvent.handle()

     *

     * If exchange needs to change interest ops, then call registerEvent() again.

     */

    void registerEvent(AsyncEvent exchange) throws IOException {

        selmgr.register(exchange);

    }

    /**

     * Only used from RawChannel to disconnect the channel from

     * the selector

     */

    void cancelRegistration(SocketChannel s) {

        selmgr.cancel(s);

    }

    /**

     * Allows an AsyncEvent to modify its interestOps.

     * @param event The modified event.

     */

    void eventUpdated(AsyncEvent event) throws ClosedChannelException {

        assert !(event instanceof AsyncTriggerEvent);

        selmgr.eventUpdated(event);

    }

    boolean isSelectorThread() {

        return Thread.currentThread() == selmgr;

    }

    Http2ClientImpl client2() {

        return client2;

    }

    private void debugCompleted(String tag, long startNanos, HttpRequest req) {

        if (debugelapsed.isLoggable(Level.DEBUG)) {

            debugelapsed.log(Level.DEBUG, () -> tag + " elapsed "

                    + (System.nanoTime() - startNanos)/1000_000L

                    + " millis for " + req.method()

                    + " to " + req.uri());

        }

    }

    @Override

    public <T> HttpResponse<T>

    send(HttpRequest req, BodyHandler<T> responseHandler)

        throws IOException, InterruptedException

    {

        try {

            return sendAsync(req, responseHandler).get();

        } catch (ExecutionException e) {

            Throwable t = e.getCause();

            if (t instanceof Error)

                throw (Error)t;

            if (t instanceof RuntimeException)

                throw (RuntimeException)t;

            else if (t instanceof IOException)

                throw Utils.getIOException(t);

            else

                throw new InternalError("Unexpected exception", t);

        }

    }

    @Override

    public <T> CompletableFuture<HttpResponse<T>>

    sendAsync(HttpRequest userRequest, BodyHandler<T> responseHandler)

    {

        AccessControlContext acc = null;

        if (System.getSecurityManager() != null)

            acc = AccessController.getContext();

        // Clone the, possibly untrusted, HttpRequest

        HttpRequestImpl requestImpl = new HttpRequestImpl(userRequest, proxySelector, acc);

        if (requestImpl.method().equals("CONNECT"))

            throw new IllegalArgumentException("Unsupported method CONNECT");

        long start = DEBUGELAPSED ? System.nanoTime() : 0;

        reference();

        try {

            debugelapsed.log(Level.DEBUG, "ClientImpl (async) send %s", userRequest);

            MultiExchange<Void,T> mex = new MultiExchange<>(userRequest,

                                                            requestImpl,

                                                            this,

                                                            responseHandler,

                                                            acc);

            CompletableFuture<HttpResponse<T>> res =

                    mex.responseAsync().whenComplete((b,t) -> unreference());

            if (DEBUGELAPSED) {

                res = res.whenComplete(

                        (b,t) -> debugCompleted("ClientImpl (async)", start, userRequest));

            }

            // makes sure that any dependent actions happen in the executor

            if (acc != null) {

                res.whenCompleteAsync((r, t) -> { /* do nothing */},

                                      new PrivilegedExecutor(executor, acc));

            }

            return res;

        } catch(Throwable t) {

            unreference();

            debugCompleted("ClientImpl (async)", start, userRequest);

            throw t;

        }

    }

    @Override

    public <U, T> CompletableFuture<U>

    sendAsync(HttpRequest userRequest, MultiSubscriber<U, T> responseHandler) {

        AccessControlContext acc = null;

        if (System.getSecurityManager() != null)

            acc = AccessController.getContext();

        // Clone the, possibly untrusted, HttpRequest

        HttpRequestImpl requestImpl = new HttpRequestImpl(userRequest, proxySelector, acc);

        if (requestImpl.method().equals("CONNECT"))

            throw new IllegalArgumentException("Unsupported method CONNECT");

        long start = DEBUGELAPSED ? System.nanoTime() : 0;

        reference();

        try {

            debugelapsed.log(Level.DEBUG, "ClientImpl (async) send multi %s", userRequest);

            MultiExchange<U,T> mex = new MultiExchange<>(userRequest,

                                                         requestImpl,

                                                         this,

                                                         responseHandler,

                                                         acc);

            CompletableFuture<U> res = mex.multiResponseAsync()

                      .whenComplete((b,t) -> unreference());

            if (DEBUGELAPSED) {

                res = res.whenComplete(

                        (b,t) -> debugCompleted("ClientImpl (async)", start, userRequest));

            }

            // makes sure that any dependent actions happen in the executor

            if (acc != null) {

                res.whenCompleteAsync((r, t) -> { /* do nothing */},

                                      new PrivilegedExecutor(executor, acc));

            }

            return res;

        } catch(Throwable t) {

            unreference();

            debugCompleted("ClientImpl (async)", start, userRequest);

            throw t;

        }