/*

 * 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 java.io.EOFException;

import java.io.IOException;

import java.io.UncheckedIOException;

import java.lang.System.Logger.Level;

import java.net.InetSocketAddress;

import java.net.URI;

import java.nio.ByteBuffer;

import java.nio.charset.StandardCharsets;

import java.util.Iterator;

import java.util.LinkedList;

import java.util.List;

import java.util.Locale;

import java.util.Map;

import java.util.Set;

import java.util.concurrent.CompletableFuture;

import java.util.ArrayList;

import java.util.Objects;

import java.util.concurrent.ConcurrentHashMap;

import java.util.concurrent.ConcurrentLinkedQueue;

import java.util.concurrent.Flow;

import java.util.function.Function;

import java.util.function.Supplier;

import javax.net.ssl.SSLEngine;

import javax.net.ssl.SSLException;

import java.net.http.HttpClient;

import java.net.http.HttpHeaders;

import jdk.internal.net.http.HttpConnection.HttpPublisher;

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

import jdk.internal.net.http.common.FlowTube.TubeSubscriber;

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

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

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

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

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

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

import jdk.internal.net.http.frame.ContinuationFrame;

import jdk.internal.net.http.frame.DataFrame;

import jdk.internal.net.http.frame.ErrorFrame;

import jdk.internal.net.http.frame.FramesDecoder;

import jdk.internal.net.http.frame.FramesEncoder;

import jdk.internal.net.http.frame.GoAwayFrame;

import jdk.internal.net.http.frame.HeaderFrame;

import jdk.internal.net.http.frame.HeadersFrame;

import jdk.internal.net.http.frame.Http2Frame;

import jdk.internal.net.http.frame.MalformedFrame;

import jdk.internal.net.http.frame.OutgoingHeaders;

import jdk.internal.net.http.frame.PingFrame;

import jdk.internal.net.http.frame.PushPromiseFrame;

import jdk.internal.net.http.frame.ResetFrame;

import jdk.internal.net.http.frame.SettingsFrame;

import jdk.internal.net.http.frame.WindowUpdateFrame;

import jdk.internal.net.http.hpack.Encoder;

import jdk.internal.net.http.hpack.Decoder;

import jdk.internal.net.http.hpack.DecodingCallback;

import static java.nio.charset.StandardCharsets.UTF_8;

import static jdk.internal.net.http.frame.SettingsFrame.*;

/**

 * An Http2Connection. Encapsulates the socket(channel) and any SSLEngine used

 * over it. Contains an HttpConnection which hides the SocketChannel SSL stuff.

 *

 * Http2Connections belong to a Http2ClientImpl, (one of) which belongs

 * to a HttpClientImpl.

 *

 * Creation cases:

 * 1) upgraded HTTP/1.1 plain tcp connection

 * 2) prior knowledge directly created plain tcp connection

 * 3) directly created HTTP/2 SSL connection which uses ALPN.

 *

 * Sending is done by writing directly to underlying HttpConnection object which

 * is operating in async mode. No flow control applies on output at this level

 * and all writes are just executed as puts to an output Q belonging to HttpConnection

 * Flow control is implemented by HTTP/2 protocol itself.

 *

 * Hpack header compression

 * and outgoing stream creation is also done here, because these operations

 * must be synchronized at the socket level. Stream objects send frames simply

 * by placing them on the connection's output Queue. sendFrame() is called

 * from a higher level (Stream) thread.

 *

 * asyncReceive(ByteBuffer) is always called from the selector thread. It assembles

 * incoming Http2Frames, and directs them to the appropriate Stream.incoming()

 * or handles them directly itself. This thread performs hpack decompression

 * and incoming stream creation (Server push). Incoming frames destined for a

 * stream are provided by calling Stream.incoming().

 */

class Http2Connection  {

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

    final static Logger DEBUG_LOGGER =

            Utils.getDebugLogger("Http2Connection"::toString, Utils.DEBUG);

    private final Logger debugHpack =

            Utils.getHpackLogger(this::dbgString, Utils.DEBUG_HPACK);

    static final ByteBuffer EMPTY_TRIGGER = ByteBuffer.allocate(0);

    static private final int MAX_CLIENT_STREAM_ID = Integer.MAX_VALUE; // 2147483647

    static private final int MAX_SERVER_STREAM_ID = Integer.MAX_VALUE - 1; // 2147483646

    /**

     * Flag set when no more streams to be opened on this connection.

     * Two cases where it is used.

     *

     * 1. Two connections to the same server were opened concurrently, in which

     *    case one of them will be put in the cache, and the second will expire

     *    when all its opened streams (which usually should be a single client

     *    stream + possibly some additional push-promise server streams) complete.

     * 2. A cached connection reaches its maximum number of streams (~ 2^31-1)

     *    either server / or client allocated, in which case it will be taken

     *    out of the cache - allowing a new connection to replace it. It will

     *    expire when all its still open streams (which could be many) eventually

     *    complete.

     */

    private boolean finalStream;

    /*

     * ByteBuffer pooling strategy for HTTP/2 protocol.

     *

     * In general there are 4 points where ByteBuffers are used:

     *  - incoming/outgoing frames from/to ByteBuffers plus incoming/outgoing

     *    encrypted data in case of SSL connection.

     *

     * 1. Outgoing frames encoded to ByteBuffers.

     *

     *  Outgoing ByteBuffers are created with required size and frequently

     *  small (except DataFrames, etc). At this place no pools at all. All

     *  outgoing buffers should eventually be collected by GC.

     *

     * 2. Incoming ByteBuffers (decoded to frames).

     *

     *  Here, total elimination of BB pool is not a good idea.

     *  We don't know how many bytes we will receive through network.

     *

     *  A possible future improvement ( currently not implemented ):

     *  Allocate buffers of reasonable size. The following life of the BB:

     *   - If all frames decoded from the BB are other than DataFrame and

     *     HeaderFrame (and HeaderFrame subclasses) BB is returned to pool,

     *   - If a DataFrame is decoded from the BB. In that case DataFrame refers

     *     to sub-buffer obtained by slice(). Such a BB is never returned to the

     *     pool and will eventually be GC'ed.

     *   - If a HeadersFrame is decoded from the BB. Then header decoding is

     *     performed inside processFrame method and the buffer could be release

     *     back to pool.

     *

     * 3. SSL encrypted buffers ( received ).

     *

     *  The current implementation recycles encrypted buffers read from the

     *  channel. The pool of buffers has a maximum size of 3, SocketTube.MAX_BUFFERS,

     *  direct buffers which are shared by all connections on a given client.

     *  The pool is used by all SSL connections - whether HTTP/1.1 or HTTP/2,

     *  but only for SSL encrypted buffers that circulate between the SocketTube

     *  Publisher and the SSLFlowDelegate Reader. Limiting the pool to this

     *  particular segment allows the use of direct buffers, thus avoiding any

     *  additional copy in the NIO socket channel implementation. See

     *  HttpClientImpl.SSLDirectBufferSupplier, SocketTube.SSLDirectBufferSource,

     *  and SSLTube.recycler.

     */

    // A small class that allows to control frames with respect to the state of

    // the connection preface. Any data received before the connection

    // preface is sent will be buffered.

    private final class FramesController {

        volatile boolean prefaceSent;

        volatile List<ByteBuffer> pending;

        boolean processReceivedData(FramesDecoder decoder, ByteBuffer buf)

                throws IOException

        {

            // if preface is not sent, buffers data in the pending list

            if (!prefaceSent) {

                if (debug.on())

                    debug.log("Preface not sent: buffering %d", buf.remaining());

                synchronized (this) {

                    if (!prefaceSent) {

                        if (pending == null) pending = new ArrayList<>();

                        pending.add(buf);

                        if (debug.on())

                            debug.log("there are now %d bytes buffered waiting for preface to be sent"

                                    + Utils.remaining(pending)

                            );

                        return false;

                    }

                }

            }

            // Preface is sent. Checks for pending data and flush it.

            // We rely on this method being called from within the Http2TubeSubscriber

            // scheduler, so we know that no other thread could execute this method

            // concurrently while we're here.

            // This ensures that later incoming buffers will not

            // be processed before we have flushed the pending queue.

            // No additional synchronization is therefore necessary here.

            List<ByteBuffer> pending = this.pending;

            this.pending = null;

            if (pending != null) {

                // flush pending data

                if (debug.on()) debug.log(() -> "Processing buffered data: "

                      + Utils.remaining(pending));

                for (ByteBuffer b : pending) {

                    decoder.decode(b);

                }

            }

            // push the received buffer to the frames decoder.

            if (buf != EMPTY_TRIGGER) {

                if (debug.on()) debug.log("Processing %d", buf.remaining());

                decoder.decode(buf);

            }

            return true;

        }

        // Mark that the connection preface is sent

        void markPrefaceSent() {

            assert !prefaceSent;

            synchronized (this) {

                prefaceSent = true;

            }

        }

    }

    volatile boolean closed;

    //-------------------------------------

    final HttpConnection connection;

    private final Http2ClientImpl client2;

    private final Map<Integer,Stream<?>> streams = new ConcurrentHashMap<>();

    private int nextstreamid;

    private int nextPushStream = 2;

    // actual stream ids are not allocated until the Headers frame is ready

    // to be sent. The following two fields are updated as soon as a stream

    // is created and assigned to a connection. They are checked before

    // assigning a stream to a connection.

    private int lastReservedClientStreamid = 1;

    private int lastReservedServerStreamid = 0;

    private int numReservedClientStreams = 0; // count of current streams

    private int numReservedServerStreams = 0; // count of current streams

    private final Encoder hpackOut;

    private final Decoder hpackIn;

    final SettingsFrame clientSettings;

    private volatile SettingsFrame serverSettings;

    private final String key; // for HttpClientImpl.connections map

    private final FramesDecoder framesDecoder;

    private final FramesEncoder framesEncoder = new FramesEncoder();

    /**

     * Send Window controller for both connection and stream windows.

     * Each of this connection's Streams MUST use this controller.

     */

    private final WindowController windowController = new WindowController();

    private final FramesController framesController = new FramesController();

    private final Http2TubeSubscriber subscriber;

    final ConnectionWindowUpdateSender windowUpdater;

    private volatile Throwable cause;

    private volatile Supplier<ByteBuffer> initial;

    static final int DEFAULT_FRAME_SIZE = 16 * 1024;

    // TODO: need list of control frames from other threads

    // that need to be sent

    private Http2Connection(HttpConnection connection,

                            Http2ClientImpl client2,

                            int nextstreamid,

                            String key) {

        this.connection = connection;

        this.client2 = client2;

        this.subscriber = new Http2TubeSubscriber(client2.client());

        this.nextstreamid = nextstreamid;

        this.key = key;

        this.clientSettings = this.client2.getClientSettings();

        this.framesDecoder = new FramesDecoder(this::processFrame,

                clientSettings.getParameter(SettingsFrame.MAX_FRAME_SIZE));

        // serverSettings will be updated by server

        this.serverSettings = SettingsFrame.getDefaultSettings();

        this.hpackOut = new Encoder(serverSettings.getParameter(HEADER_TABLE_SIZE));

        this.hpackIn = new Decoder(clientSettings.getParameter(HEADER_TABLE_SIZE));

        if (debugHpack.on()) {

            debugHpack.log("For the record:" + super.toString());

            debugHpack.log("Decoder created: %s", hpackIn);

            debugHpack.log("Encoder created: %s", hpackOut);

        }

        this.windowUpdater = new ConnectionWindowUpdateSender(this,

                client2.getConnectionWindowSize(clientSettings));

    }

    /**

     * Case 1) Create from upgraded HTTP/1.1 connection.

     * Is ready to use. Can't be SSL. exchange is the Exchange

     * that initiated the connection, whose response will be delivered

     * on a Stream.

     */

    private Http2Connection(HttpConnection connection,

                    Http2ClientImpl client2,

                    Exchange<?> exchange,

                    Supplier<ByteBuffer> initial)

        throws IOException, InterruptedException

    {

        this(connection,

                client2,

                3, // stream 1 is registered during the upgrade

                keyFor(connection));

        reserveStream(true);

        Log.logTrace("Connection send window size {0} ", windowController.connectionWindowSize());

        Stream<?> initialStream = createStream(exchange);

        initialStream.registerStream(1);

        windowController.registerStream(1, getInitialSendWindowSize());

        initialStream.requestSent();

        // Upgrading:

        //    set callbacks before sending preface - makes sure anything that

        //    might be sent by the server will come our way.

        this.initial = initial;

        connectFlows(connection);

        sendConnectionPreface();

    }

    // Used when upgrading an HTTP/1.1 connection to HTTP/2 after receiving

    // agreement from the server. Async style but completes immediately, because

    // the connection is already connected.

    static CompletableFuture<Http2Connection> createAsync(HttpConnection connection,

                                                          Http2ClientImpl client2,

                                                          Exchange<?> exchange,

                                                          Supplier<ByteBuffer> initial)

    {

        return MinimalFuture.supply(() -> new Http2Connection(connection, client2, exchange, initial));

    }

    // Requires TLS handshake. So, is really async

    static CompletableFuture<Http2Connection> createAsync(HttpRequestImpl request,

                                                          Http2ClientImpl h2client) {

        assert request.secure();

        AbstractAsyncSSLConnection connection = (AbstractAsyncSSLConnection)

        HttpConnection.getConnection(request.getAddress(),

                                     h2client.client(),

                                     request,

                                     HttpClient.Version.HTTP_2);

        return connection.connectAsync()

                  .thenCompose(unused -> checkSSLConfig(connection))

                  .thenCompose(notused-> {

                      CompletableFuture<Http2Connection> cf = new MinimalFuture<>();

                      try {

                          Http2Connection hc = new Http2Connection(request, h2client, connection);

                          cf.complete(hc);

                      } catch (IOException e) {

                          cf.completeExceptionally(e);

                      }

                      return cf; } );

    }

    /**

     * Cases 2) 3)

     *

     * request is request to be sent.

     */

    private Http2Connection(HttpRequestImpl request,

                            Http2ClientImpl h2client,

                            HttpConnection connection)

        throws IOException

    {

        this(connection,

             h2client,

             1,

             keyFor(request.uri(), request.proxy()));

        Log.logTrace("Connection send window size {0} ", windowController.connectionWindowSize());

        // safe to resume async reading now.

        connectFlows(connection);

        sendConnectionPreface();

    }

    private void connectFlows(HttpConnection connection) {

        FlowTube tube =  connection.getConnectionFlow();

        // Connect the flow to our Http2TubeSubscriber:

        tube.connectFlows(connection.publisher(), subscriber);

    }

    final HttpClientImpl client() {

        return client2.client();

    }

    // call these before assigning a request/stream to a connection

    // if false returned then a new Http2Connection is required

    // if true, the the stream may be assigned to this connection

    // for server push, if false returned, then the stream should be cancelled

    synchronized boolean reserveStream(boolean clientInitiated) throws IOException {

        if (finalStream) {

            return false;

        }

        if (clientInitiated && (lastReservedClientStreamid + 2) >= MAX_CLIENT_STREAM_ID) {

            setFinalStream();

            client2.deleteConnection(this);

            return false;

        } else if (!clientInitiated && (lastReservedServerStreamid + 2) >= MAX_SERVER_STREAM_ID) {

            setFinalStream();

            client2.deleteConnection(this);

            return false;

        }

        if (clientInitiated)

            lastReservedClientStreamid+=2;

        else

            lastReservedServerStreamid+=2;

        assert numReservedClientStreams >= 0;

        assert numReservedServerStreams >= 0;

        if (clientInitiated && numReservedClientStreams >= getMaxConcurrentClientStreams()) {

            throw new IOException("too many concurrent streams");

        } else if (clientInitiated) {

            numReservedClientStreams++;

        }

        if (!clientInitiated && numReservedServerStreams >= getMaxConcurrentServerStreams()) {

            return false;

        } else if (!clientInitiated) {

            numReservedServerStreams++;

        }

        return true;

    }

    /**

     * Throws an IOException if h2 was not negotiated

     */

    private static CompletableFuture<?> checkSSLConfig(AbstractAsyncSSLConnection aconn) {

        assert aconn.isSecure();

        Function<String, CompletableFuture<Void>> checkAlpnCF = (alpn) -> {

            CompletableFuture<Void> cf = new MinimalFuture<>();

            SSLEngine engine = aconn.getEngine();

            assert Objects.equals(alpn, engine.getApplicationProtocol());

            DEBUG_LOGGER.log("checkSSLConfig: alpn: %s", alpn );

            if (alpn == null || !alpn.equals("h2")) {

                String msg;

                if (alpn == null) {

                    Log.logSSL("ALPN not supported");

                    msg = "ALPN not supported";

                } else {

                    switch (alpn) {

                        case "":

                            Log.logSSL(msg = "No ALPN negotiated");

                            break;

                        case "http/1.1":

                            Log.logSSL( msg = "HTTP/1.1 ALPN returned");

                            break;

                        default:

                            Log.logSSL(msg = "Unexpected ALPN: " + alpn);

                            cf.completeExceptionally(new IOException(msg));

                    }

                }

                cf.completeExceptionally(new ALPNException(msg, aconn));

                return cf;

            }

            cf.complete(null);

            return cf;

        };

        return aconn.getALPN()

                .whenComplete((r,t) -> {

                    if (t != null && t instanceof SSLException) {

                        // something went wrong during the initial handshake

                        // close the connection

                        aconn.close();

                    }

                })

                .thenCompose(checkAlpnCF);

    }

    synchronized boolean finalStream() {

        return finalStream;