/*

 * 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.incubator.http;

import java.io.IOException;

import java.lang.System.Logger.Level;

import java.net.URI;

import java.nio.ByteBuffer;

import java.util.ArrayList;

import java.util.Collections;

import java.util.List;

import java.util.Optional;

import java.util.concurrent.CompletableFuture;

import java.util.concurrent.ConcurrentLinkedDeque;

import java.util.concurrent.ConcurrentLinkedQueue;

import java.util.concurrent.Executor;

import java.util.concurrent.Flow;

import java.util.concurrent.Flow.Subscription;

import java.util.concurrent.atomic.AtomicReference;

import jdk.incubator.http.HttpResponse.BodySubscriber;

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

import jdk.incubator.http.internal.frame.*;

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

/**

 * Http/2 Stream handling.

 *

 * REQUESTS

 *

 * sendHeadersOnly() -- assembles HEADERS frame and puts on connection outbound Q

 *

 * sendRequest() -- sendHeadersOnly() + sendBody()

 *

 * sendBodyAsync() -- calls sendBody() in an executor thread.

 *

 * sendHeadersAsync() -- calls sendHeadersOnly() which does not block

 *

 * sendRequestAsync() -- calls sendRequest() in an executor thread

 *

 * RESPONSES

 *

 * Multiple responses can be received per request. Responses are queued up on

 * a LinkedList of CF<HttpResponse> and the the first one on the list is completed

 * with the next response

 *

 * getResponseAsync() -- queries list of response CFs and returns first one

 *               if one exists. Otherwise, creates one and adds it to list

 *               and returns it. Completion is achieved through the

 *               incoming() upcall from connection reader thread.

 *

 * getResponse() -- calls getResponseAsync() and waits for CF to complete

 *

 * responseBodyAsync() -- calls responseBody() in an executor thread.

 *

 * incoming() -- entry point called from connection reader thread. Frames are

 *               either handled immediately without blocking or for data frames

 *               placed on the stream's inputQ which is consumed by the stream's

 *               reader thread.

 *

 * PushedStream sub class

 * ======================

 * Sending side methods are not used because the request comes from a PUSH_PROMISE

 * frame sent by the server. When a PUSH_PROMISE is received the PushedStream

 * is created. PushedStream does not use responseCF list as there can be only

 * one response. The CF is created when the object created and when the response

 * HEADERS frame is received the object is completed.

 */

class Stream<T> extends ExchangeImpl<T> {

    final static boolean DEBUG = Utils.DEBUG; // Revisit: temporary developer's flag

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

    final ConcurrentLinkedQueue<Http2Frame> inputQ = new ConcurrentLinkedQueue<>();

    final SequentialScheduler sched =

            SequentialScheduler.synchronizedScheduler(this::schedule);

    final SubscriptionBase userSubscription = new SubscriptionBase(sched, this::cancel);

    /**

     * This stream's identifier. Assigned lazily by the HTTP2Connection before

     * the stream's first frame is sent.

     */

    protected volatile int streamid;

    long requestContentLen;

    final Http2Connection connection;

    final HttpRequestImpl request;

    final DecodingCallback rspHeadersConsumer;

    HttpHeadersImpl responseHeaders;

    final HttpHeadersImpl requestPseudoHeaders;

    volatile HttpResponse.BodySubscriber<T> responseSubscriber;

    final HttpRequest.BodyPublisher requestPublisher;

    volatile RequestSubscriber requestSubscriber;

    volatile int responseCode;

    volatile Response response;

    volatile Throwable failed; // The exception with which this stream was canceled.

    final CompletableFuture<Void> requestBodyCF = new MinimalFuture<>();

    volatile CompletableFuture<T> responseBodyCF;

    /** True if END_STREAM has been seen in a frame received on this stream. */

    private volatile boolean remotelyClosed;

    private volatile boolean closed;

    private volatile boolean endStreamSent;

    // state flags

    private boolean requestSent, responseReceived;

    /**

     * A reference to this Stream's connection Send Window controller. The

     * stream MUST acquire the appropriate amount of Send Window before

     * sending any data. Will be null for PushStreams, as they cannot send data.

     */

    private final WindowController windowController;

    private final WindowUpdateSender windowUpdater;

    @Override

    HttpConnection connection() {

        return connection.connection;

    }

    /**

     * Invoked either from incoming() -> {receiveDataFrame() or receiveResetFrame() }

     * of after user subscription window has re-opened, from SubscriptionBase.request()

     */

    private void schedule() {

        if (responseSubscriber == null)

            // can't process anything yet

            return;

        while (!inputQ.isEmpty()) {

            Http2Frame frame  = inputQ.peek();

            if (frame instanceof ResetFrame) {

                inputQ.remove();

                handleReset((ResetFrame)frame);

                return;

            }

            DataFrame df = (DataFrame)frame;

            boolean finished = df.getFlag(DataFrame.END_STREAM);

            List<ByteBuffer> buffers = df.getData();

            List<ByteBuffer> dsts = Collections.unmodifiableList(buffers);

            int size = Utils.remaining(dsts, Integer.MAX_VALUE);

            if (size == 0 && finished) {

                inputQ.remove();

                Log.logTrace("responseSubscriber.onComplete");

                debug.log(Level.DEBUG, "incoming: onComplete");

                sched.stop();

                responseSubscriber.onComplete();

                setEndStreamReceived();

                return;

            } else if (userSubscription.tryDecrement()) {

                inputQ.remove();

                Log.logTrace("responseSubscriber.onNext {0}", size);

                debug.log(Level.DEBUG, "incoming: onNext(%d)", size);

                responseSubscriber.onNext(dsts);

                if (consumed(df)) {

                    Log.logTrace("responseSubscriber.onComplete");

                    debug.log(Level.DEBUG, "incoming: onComplete");

                    sched.stop();

                    responseSubscriber.onComplete();

                    setEndStreamReceived();

                    return;

                }

            } else {

                return;

            }

        }

        Throwable t = failed;

        if (t != null) {

            sched.stop();

            responseSubscriber.onError(t);

            close();

        }

    }

    // Callback invoked after the Response BodySubscriber has consumed the

    // buffers contained in a DataFrame.

    // Returns true if END_STREAM is reached, false otherwise.

    private boolean consumed(DataFrame df) {

        // RFC 7540 6.1:

        // The entire DATA frame payload is included in flow control,

        // including the Pad Length and Padding fields if present

        int len = df.payloadLength();

        connection.windowUpdater.update(len);

        if (!df.getFlag(DataFrame.END_STREAM)) {

            // Don't send window update on a stream which is

            // closed or half closed.

            windowUpdater.update(len);

            return false; // more data coming

        }

        return true; // end of stream

    }

    @Override

    CompletableFuture<T> readBodyAsync(HttpResponse.BodyHandler<T> handler,

                                       boolean returnConnectionToPool,

                                       Executor executor)

    {

        Log.logTrace("Reading body on stream {0}", streamid);

        BodySubscriber<T> bodySubscriber = handler.apply(responseCode, responseHeaders);

        CompletableFuture<T> cf = receiveData(bodySubscriber);

        PushGroup<?> pg = exchange.getPushGroup();

        if (pg != null) {

            // if an error occurs make sure it is recorded in the PushGroup

            cf = cf.whenComplete((t,e) -> pg.pushError(e));

        }

        return cf;

    }

    @Override

    public String toString() {

        StringBuilder sb = new StringBuilder();

        sb.append("streamid: ")

                .append(streamid);

        return sb.toString();

    }

    private void receiveDataFrame(DataFrame df) {

        inputQ.add(df);

        sched.runOrSchedule();

    }

    /** Handles a RESET frame. RESET is always handled inline in the queue. */

    private void receiveResetFrame(ResetFrame frame) {

        inputQ.add(frame);

        sched.runOrSchedule();

    }

    // pushes entire response body into response subscriber

    // blocking when required by local or remote flow control

    CompletableFuture<T> receiveData(BodySubscriber<T> bodySubscriber) {

        responseBodyCF = MinimalFuture.of(bodySubscriber.getBody());

        if (isCanceled()) {

            Throwable t = getCancelCause();

            responseBodyCF.completeExceptionally(t);

        } else {

            bodySubscriber.onSubscribe(userSubscription);

        }

        // Set the responseSubscriber field now that onSubscribe has been called.

        // This effectively allows the scheduler to start invoking the callbacks.

        responseSubscriber = bodySubscriber;

        sched.runOrSchedule(); // in case data waiting already to be processed

        return responseBodyCF;

    }

    @Override

    CompletableFuture<ExchangeImpl<T>> sendBodyAsync() {

        return sendBodyImpl().thenApply( v -> this);

    }

    @SuppressWarnings("unchecked")

    Stream(Http2Connection connection,

           Exchange<T> e,

           WindowController windowController)

    {

        super(e);

        this.connection = connection;

        this.windowController = windowController;

        this.request = e.request();

        this.requestPublisher = request.requestPublisher;  // may be null

        responseHeaders = new HttpHeadersImpl();

        rspHeadersConsumer = (name, value) -> {

            responseHeaders.addHeader(name.toString(), value.toString());

            if (Log.headers() && Log.trace()) {

                Log.logTrace("RECEIVED HEADER (streamid={0}): {1}: {2}",

                             streamid, name, value);

            }

        };

        this.requestPseudoHeaders = new HttpHeadersImpl();

        // NEW

        this.windowUpdater = new StreamWindowUpdateSender(connection);

    }

    /**

     * Entry point from Http2Connection reader thread.

     *

     * Data frames will be removed by response body thread.

     */

    void incoming(Http2Frame frame) throws IOException {

        debug.log(Level.DEBUG, "incoming: %s", frame);

        if ((frame instanceof HeaderFrame)) {

            HeaderFrame hframe = (HeaderFrame)frame;

            if (hframe.endHeaders()) {

                Log.logTrace("handling response (streamid={0})", streamid);

                handleResponse();

                if (hframe.getFlag(HeaderFrame.END_STREAM)) {

                    receiveDataFrame(new DataFrame(streamid, DataFrame.END_STREAM, List.of()));

                }

            }

        } else if (frame instanceof DataFrame) {

            receiveDataFrame((DataFrame)frame);

        } else {

            otherFrame(frame);

        }

    }

    void otherFrame(Http2Frame frame) throws IOException {

        switch (frame.type()) {

            case WindowUpdateFrame.TYPE:

                incoming_windowUpdate((WindowUpdateFrame) frame);

                break;

            case ResetFrame.TYPE:

                incoming_reset((ResetFrame) frame);

                break;

            case PriorityFrame.TYPE:

                incoming_priority((PriorityFrame) frame);

                break;

            default:

                String msg = "Unexpected frame: " + frame.toString();

                throw new IOException(msg);

        }

    }

    // The Hpack decoder decodes into one of these consumers of name,value pairs

    DecodingCallback rspHeadersConsumer() {

        return rspHeadersConsumer;

    }

    protected void handleResponse() throws IOException {

        responseCode = (int)responseHeaders

                .firstValueAsLong(":status")

                .orElseThrow(() -> new IOException("no statuscode in response"));

        response = new Response(

                request, exchange, responseHeaders,

                responseCode, HttpClient.Version.HTTP_2);

        /* TODO: review if needs to be removed

           the value is not used, but in case `content-length` doesn't parse as

           long, there will be NumberFormatException. If left as is, make sure

           code up the stack handles NFE correctly. */

        responseHeaders.firstValueAsLong("content-length");

        if (Log.headers()) {

            StringBuilder sb = new StringBuilder("RESPONSE HEADERS:\n");

            Log.dumpHeaders(sb, "    ", responseHeaders);

            Log.logHeaders(sb.toString());

        }

        completeResponse(response);

    }

    void incoming_reset(ResetFrame frame) {

        Log.logTrace("Received RST_STREAM on stream {0}", streamid);

        if (endStreamReceived()) {

            Log.logTrace("Ignoring RST_STREAM frame received on remotely closed stream {0}", streamid);

        } else if (closed) {

            Log.logTrace("Ignoring RST_STREAM frame received on closed stream {0}", streamid);

        } else {

            // put it in the input queue in order to read all

            // pending data frames first. Indeed, a server may send

            // RST_STREAM after sending END_STREAM, in which case we should

            // ignore it. However, we won't know if we have received END_STREAM

            // or not until all pending data frames are read.

            receiveResetFrame(frame);

            // RST_STREAM was pushed to the queue. It will be handled by

            // asyncReceive after all pending data frames have been

            // processed.

            Log.logTrace("RST_STREAM pushed in queue for stream {0}", streamid);

        }

    }

    void handleReset(ResetFrame frame) {

        Log.logTrace("Handling RST_STREAM on stream {0}", streamid);

        if (!closed) {

            close();

            int error = frame.getErrorCode();

            completeResponseExceptionally(new IOException(ErrorFrame.stringForCode(error)));

        } else {

            Log.logTrace("Ignoring RST_STREAM frame received on closed stream {0}", streamid);

        }

    }

    void incoming_priority(PriorityFrame frame) {

        // TODO: implement priority

        throw new UnsupportedOperationException("Not implemented");

    }

    private void incoming_windowUpdate(WindowUpdateFrame frame)

        throws IOException

    {

        int amount = frame.getUpdate();

        if (amount <= 0) {

            Log.logTrace("Resetting stream: {0} %d, Window Update amount: %d\n",

                         streamid, streamid, amount);

            connection.resetStream(streamid, ResetFrame.FLOW_CONTROL_ERROR);

        } else {

            assert streamid != 0;

            boolean success = windowController.increaseStreamWindow(amount, streamid);

            if (!success) {  // overflow

                connection.resetStream(streamid, ResetFrame.FLOW_CONTROL_ERROR);

            }

        }

    }

    void incoming_pushPromise(HttpRequestImpl pushRequest,

                              PushedStream<T> pushStream)

        throws IOException

    {

        if (Log.requests()) {

            Log.logRequest("PUSH_PROMISE: " + pushRequest.toString());

        }

        PushGroup<T> pushGroup = exchange.getPushGroup();

        if (pushGroup == null) {

            Log.logTrace("Rejecting push promise stream " + streamid);

            connection.resetStream(pushStream.streamid, ResetFrame.REFUSED_STREAM);

            pushStream.close();

            return;

        }

        PushGroup.Acceptor<T> acceptor = pushGroup.acceptPushRequest(pushRequest);

        if (!acceptor.accepted()) {

            // cancel / reject

            IOException ex = new IOException("Stream " + streamid + " cancelled by users handler");

            if (Log.trace()) {

                Log.logTrace("No body subscriber for {0}: {1}", pushRequest,

                        ex.getMessage());

            }

            pushStream.cancelImpl(ex);

            return;

        }

        CompletableFuture<HttpResponse<T>> pushResponseCF = acceptor.cf();

        HttpResponse.BodyHandler<T> pushHandler = acceptor.bodyHandler();

        assert pushHandler != null;

        pushStream.requestSent();

        pushStream.setPushHandler(pushHandler);  // TODO: could wrap the handler to throw on acceptPushPromise ?

        // setup housekeeping for when the push is received

        // TODO: deal with ignoring of CF anti-pattern

        CompletableFuture<HttpResponse<T>> cf = pushStream.responseCF();

        cf.whenComplete((HttpResponse<T> resp, Throwable t) -> {

            t = Utils.getCompletionCause(t);

            if (Log.trace()) {

                Log.logTrace("Push completed on stream {0} for {1}{2}",

                             pushStream.streamid, resp,

                             ((t==null) ? "": " with exception " + t));

            }

            if (t != null) {

                pushGroup.pushError(t);

                pushResponseCF.completeExceptionally(t);

            } else {

                pushResponseCF.complete(resp);

            }

            pushGroup.pushCompleted();

        });

    }

    private OutgoingHeaders<Stream<T>> headerFrame(long contentLength) {

        HttpHeadersImpl h = request.getSystemHeaders();

        if (contentLength > 0) {

            h.setHeader("content-length", Long.toString(contentLength));

        }

        setPseudoHeaderFields();

        OutgoingHeaders<Stream<T>> f = new OutgoingHeaders<>(h, request.getUserHeaders(), this);

        if (contentLength == 0) {

            f.setFlag(HeadersFrame.END_STREAM);

            endStreamSent = true;

        }

        return f;

    }

    private void setPseudoHeaderFields() {

        HttpHeadersImpl hdrs = requestPseudoHeaders;

        String method = request.method();

        hdrs.setHeader(":method", method);

        URI uri = request.uri();

        hdrs.setHeader(":scheme", uri.getScheme());

        // TODO: userinfo deprecated. Needs to be removed

        hdrs.setHeader(":authority", uri.getAuthority());

        // TODO: ensure header names beginning with : not in user headers

        String query = uri.getQuery();

        String path = uri.getPath();

        if (path == null || path.isEmpty()) {

            if (method.equalsIgnoreCase("OPTIONS")) {

                path = "*";

            } else {

                path = "/";

            }

        }

        if (query != null) {

            path += "?" + query;