/*

 * Copyright (c) 2015, 2019, 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.net.URI;

import java.nio.ByteBuffer;

import java.util.ArrayList;

import java.util.Collections;

import java.util.List;

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.AtomicBoolean;

import java.util.concurrent.atomic.AtomicReference;

import java.util.function.BiPredicate;

import java.net.http.HttpClient;

import java.net.http.HttpHeaders;

import java.net.http.HttpRequest;

import java.net.http.HttpResponse;

import java.net.http.HttpResponse.BodySubscriber;

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

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

import jdk.internal.net.http.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 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 Logger debug = Utils.getDebugLogger(this::dbgString, Utils.DEBUG);

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

    final SequentialScheduler sched =

            SequentialScheduler.synchronizedScheduler(this::schedule);

    final SubscriptionBase userSubscription =

            new SubscriptionBase(sched, this::cancel, this::onSubscriptionError);

    /**

     * 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 HeadersConsumer rspHeadersConsumer;

    final HttpHeadersBuilder responseHeadersBuilder;

    final HttpHeaders requestPseudoHeaders;

    volatile HttpResponse.BodySubscriber<T> responseSubscriber;

    final HttpRequest.BodyPublisher requestPublisher;

    volatile RequestSubscriber requestSubscriber;

    volatile int responseCode;

    volatile Response response;

    // The exception with which this stream was canceled.

    private final AtomicReference<Throwable> errorRef = new AtomicReference<>();

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

    volatile CompletableFuture<T> responseBodyCF;

    volatile HttpResponse.BodySubscriber<T> pendingResponseSubscriber;

    volatile boolean stopRequested;

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

    final AtomicBoolean deRegistered = new AtomicBoolean(false);

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

        boolean onCompleteCalled = false;

        HttpResponse.BodySubscriber<T> subscriber = responseSubscriber;

        try {

            if (subscriber == null) {

                subscriber = responseSubscriber = pendingResponseSubscriber;

                if (subscriber == null) {

                    // can't process anything yet

                    return;

                } else {

                    if (debug.on()) debug.log("subscribing user subscriber");

                    subscriber.onSubscribe(userSubscription);

                }

            }

            while (!inputQ.isEmpty()) {

                Http2Frame frame = inputQ.peek();

                if (frame instanceof ResetFrame) {

                    inputQ.remove();

                    handleReset((ResetFrame)frame, subscriber);

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

                    connection.ensureWindowUpdated(df); // must update connection window

                    Log.logTrace("responseSubscriber.onComplete");

                    if (debug.on()) debug.log("incoming: onComplete");

                    sched.stop();

                    connection.decrementStreamsCount(streamid);

                    subscriber.onComplete();

                    onCompleteCalled = true;

                    setEndStreamReceived();

                    return;

                } else if (userSubscription.tryDecrement()) {

                    inputQ.remove();

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

                    if (debug.on()) debug.log("incoming: onNext(%d)", size);

                    try {

                        subscriber.onNext(dsts);

                    } catch (Throwable t) {

                        connection.dropDataFrame(df); // must update connection window

                        throw t;

                    }

                    if (consumed(df)) {

                        Log.logTrace("responseSubscriber.onComplete");

                        if (debug.on()) debug.log("incoming: onComplete");

                        sched.stop();

                        connection.decrementStreamsCount(streamid);

                        subscriber.onComplete();

                        onCompleteCalled = true;

                        setEndStreamReceived();

                        return;

                    }

                } else {

                    if (stopRequested) break;

                    return;

                }

            }

        } catch (Throwable throwable) {

            errorRef.compareAndSet(null, throwable);

        } finally {

            if (sched.isStopped()) drainInputQueue();

        }

        Throwable t = errorRef.get();

        if (t != null) {

            sched.stop();

            try {

                if (!onCompleteCalled) {

                    if (debug.on())

                        debug.log("calling subscriber.onError: %s", (Object) t);

                    subscriber.onError(t);

                } else {

                    if (debug.on())

                        debug.log("already completed: dropping error %s", (Object) t);

                }

            } catch (Throwable x) {

                Log.logError("Subscriber::onError threw exception: {0}", (Object) t);

            } finally {

                cancelImpl(t);

                drainInputQueue();

            }

        }

    }

    // must only be called from the scheduler schedule() loop.

    // ensure that all received data frames are accounted for

    // in the connection window flow control if the scheduler

    // is stopped before all the data is consumed.

    private void drainInputQueue() {

        Http2Frame frame;

        while ((frame = inputQ.poll()) != null) {

            if (frame instanceof DataFrame) {

                connection.dropDataFrame((DataFrame)frame);

            }

        }

    }

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

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

        if (len == 0) return endStream;

        connection.windowUpdater.update(len);

        if (!endStream) {

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

            // closed or half closed.

            windowUpdater.update(len);

        }

        // true: end of stream; false: more data coming

        return endStream;

    }

    boolean deRegister() {

        return deRegistered.compareAndSet(false, true);

    }

    @Override

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

                                       boolean returnConnectionToPool,

                                       Executor executor)

    {

        try {

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

            debug.log("Getting BodySubscriber for: " + response);

            BodySubscriber<T> bodySubscriber = handler.apply(new ResponseInfoImpl(response));

            CompletableFuture<T> cf = receiveData(bodySubscriber, executor);

            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;

        } catch (Throwable t) {

            // may be thrown by handler.apply

            cancelImpl(t);

            return MinimalFuture.failedFuture(t);

        }

    }

    @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, Executor executor) {

        // We want to allow the subscriber's getBody() method to block so it

        // can work with InputStreams. So, we offload execution.

        responseBodyCF = ResponseSubscribers.getBodyAsync(executor, bodySubscriber,

                new MinimalFuture<>(), this::cancelImpl);

        if (isCanceled()) {

            Throwable t = getCancelCause();

            responseBodyCF.completeExceptionally(t);

        } else {

            pendingResponseSubscriber = 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

        this.responseHeadersBuilder = new HttpHeadersBuilder();

        this.rspHeadersConsumer = new HeadersConsumer();

        this.requestPseudoHeaders = createPseudoHeaders(request);

        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 {

        if (debug.on()) debug.log("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 {

        HttpHeaders responseHeaders = responseHeadersBuilder.build();

        responseCode = (int)responseHeaders

                .firstValueAsLong(":status")

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

        response = new Response(

                request, exchange, responseHeaders, connection(),

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

        }

        // this will clear the response headers

        rspHeadersConsumer.reset();

        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 {

            Flow.Subscriber<?> subscriber =

                    responseSubscriber == null ? pendingResponseSubscriber : responseSubscriber;

            if (response == null && subscriber == null) {

                // we haven't receive the headers yet, and won't receive any!

                // handle reset now.

                handleReset(frame, subscriber);

            } 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, Flow.Subscriber<?> subscriber) {

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

        if (!closed) {

            synchronized (this) {

                if (closed) {

                    if (debug.on()) debug.log("Stream already closed: ignoring RESET");

                    return;

                }

                closed = true;

            }

            try {

                int error = frame.getErrorCode();

                IOException e = new IOException("Received RST_STREAM: "

                        + ErrorFrame.stringForCode(error));

                if (errorRef.compareAndSet(null, e)) {

                    if (subscriber != null) {

                        subscriber.onError(e);

                    }

                }

                completeResponseExceptionally(e);

                if (!requestBodyCF.isDone()) {

                    requestBodyCF.completeExceptionally(errorRef.get()); // we may be sending the body..

                }

                if (responseBodyCF != null) {

                    responseBodyCF.completeExceptionally(errorRef.get());

                }

            } finally {

                connection.decrementStreamsCount(streamid);