/*

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

import java.lang.System.Logger.Level;

import java.nio.ByteBuffer;

import java.util.concurrent.CompletableFuture;

import java.util.concurrent.CompletionStage;

import java.util.concurrent.Executor;

import java.util.function.BiConsumer;

import java.util.function.Consumer;

import java.util.function.Function;

import jdk.incubator.http.ResponseContent.BodyParser;

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

import static jdk.incubator.http.HttpClient.Version.HTTP_1_1;

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

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

/**

 * Handles a HTTP/1.1 response (headers + body).

 * There can be more than one of these per Http exchange.

 */

class Http1Response<T> {

    private volatile ResponseContent content;

    private final HttpRequestImpl request;

    private Response response;

    private final HttpConnection connection;

    private HttpHeaders headers;

    private int responseCode;

    private final Http1Exchange<T> exchange;

    private boolean return2Cache; // return connection to cache when finished

    private final HeadersReader headersReader; // used to read the headers

    private final BodyReader bodyReader; // used to read the body

    private final Http1AsyncReceiver asyncReceiver;

    private volatile EOFException eof;

    // max number of bytes of (fixed length) body to ignore on redirect

    private final static int MAX_IGNORE = 1024;

    // Revisit: can we get rid of this?

    static enum State {INITIAL, READING_HEADERS, READING_BODY, DONE}

    private volatile State readProgress = State.INITIAL;

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

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

    Http1Response(HttpConnection conn,

                  Http1Exchange<T> exchange,

                  Http1AsyncReceiver asyncReceiver) {

        this.readProgress = State.INITIAL;

        this.request = exchange.request();

        this.exchange = exchange;

        this.connection = conn;

        this.asyncReceiver = asyncReceiver;

        headersReader = new HeadersReader(this::advance);

        bodyReader = new BodyReader(this::advance);

    }

   public CompletableFuture<Response> readHeadersAsync(Executor executor) {

        debug.log(Level.DEBUG, () -> "Reading Headers: (remaining: "

                + asyncReceiver.remaining() +") "  + readProgress);

        // with expect continue we will resume reading headers + body.

        asyncReceiver.unsubscribe(bodyReader);

        bodyReader.reset();

        Http1HeaderParser hd = new Http1HeaderParser();

        readProgress = State.READING_HEADERS;

        headersReader.start(hd);

        asyncReceiver.subscribe(headersReader);

        CompletableFuture<State> cf = headersReader.completion();

        assert cf != null : "parsing not started";

        Function<State, Response> lambda = (State completed) -> {

                assert completed == State.READING_HEADERS;

                debug.log(Level.DEBUG, () ->

                            "Reading Headers: creating Response object;"

                            + " state is now " + readProgress);

                asyncReceiver.unsubscribe(headersReader);

                responseCode = hd.responseCode();

                headers = hd.headers();

                response = new Response(request,

                                        exchange.getExchange(),

                                        headers,

                                        responseCode,

                                        HTTP_1_1);

                return response;

            };

        if (executor != null) {

            return cf.thenApplyAsync(lambda, executor);

        } else {

            return cf.thenApply(lambda);

        }

    }

    private boolean finished;

    synchronized void completed() {

        finished = true;

    }

    synchronized boolean finished() {

        return finished;

    }

    int fixupContentLen(int clen) {

        if (request.method().equalsIgnoreCase("HEAD")) {

            return 0;

        }

        if (clen == -1) {

            if (headers.firstValue("Transfer-encoding").orElse("")

                       .equalsIgnoreCase("chunked")) {

                return -1;

            }

            return 0;

        }

        return clen;

    }

    /**

     * Read up to MAX_IGNORE bytes discarding

     */

    public CompletableFuture<Void> ignoreBody(Executor executor) {

        int clen = (int)headers.firstValueAsLong("Content-Length").orElse(-1);

        if (clen == -1 || clen > MAX_IGNORE) {

            connection.close();

            return MinimalFuture.completedFuture(null); // not treating as error

        } else {

            return readBody(HttpResponse.BodySubscriber.discard((Void)null), true, executor);

        }

    }

    public <U> CompletableFuture<U> readBody(HttpResponse.BodySubscriber<U> p,

                                         boolean return2Cache,

                                         Executor executor) {

        this.return2Cache = return2Cache;

        final HttpResponse.BodySubscriber<U> pusher = p;

        final CompletionStage<U> bodyCF = p.getBody();

        final CompletableFuture<U> cf = MinimalFuture.of(bodyCF);

        int clen0 = (int)headers.firstValueAsLong("Content-Length").orElse(-1);

        final int clen = fixupContentLen(clen0);

        // expect-continue reads headers and body twice.

        // if we reach here, we must reset the headersReader state.

        asyncReceiver.unsubscribe(headersReader);

        headersReader.reset();

        executor.execute(() -> {

            try {

                HttpClientImpl client = connection.client();

                content = new ResponseContent(

                        connection, clen, headers, pusher,

                        this::onFinished

                );

                if (cf.isCompletedExceptionally()) {

                    // if an error occurs during subscription

                    connection.close();

                    return;

                }

                // increment the reference count on the HttpClientImpl

                // to prevent the SelectorManager thread from exiting until

                // the body is fully read.

                client.reference();

                bodyReader.start(content.getBodyParser(

                    (t) -> {

                        try {

                            if (t != null) {

                                pusher.onError(t);

                                connection.close();

                                if (!cf.isDone())

                                    cf.completeExceptionally(t);

                            }

                        } finally {

                            // decrement the reference count on the HttpClientImpl

                            // to allow the SelectorManager thread to exit if no

                            // other operation is pending and the facade is no

                            // longer referenced.

                            client.unreference();

                            bodyReader.onComplete(t);

                        }

                    }));

                CompletableFuture<State> bodyReaderCF = bodyReader.completion();

                asyncReceiver.subscribe(bodyReader);

                assert bodyReaderCF != null : "parsing not started";

                // Make sure to keep a reference to asyncReceiver from

                // within this

                CompletableFuture<?> trailingOp = bodyReaderCF.whenComplete((s,t) ->  {

                    t = Utils.getCompletionCause(t);

                    try {

                        if (t != null) {

                            debug.log(Level.DEBUG, () ->

                                    "Finished reading body: " + s);

                            assert s == State.READING_BODY;

                        }

                        if (t != null && !cf.isDone()) {

                            pusher.onError(t);

                            cf.completeExceptionally(t);

                        }

                    } catch (Throwable x) {

                        // not supposed to happen

                        asyncReceiver.onReadError(x);

                    }

                });

                connection.addTrailingOperation(trailingOp);

            } catch (Throwable t) {

               debug.log(Level.DEBUG, () -> "Failed reading body: " + t);

                try {

                    if (!cf.isDone()) {

                        pusher.onError(t);

                        cf.completeExceptionally(t);

                    }

                } finally {

                    asyncReceiver.onReadError(t);

                }

            }

        });

        return cf;

    }

    private void onFinished() {

        asyncReceiver.clear();

        if (return2Cache) {

            Log.logTrace("Attempting to return connection to the pool: {0}", connection);

            // TODO: need to do something here?

            // connection.setAsyncCallbacks(null, null, null);

            // don't return the connection to the cache if EOF happened.

            debug.log(Level.DEBUG, () -> connection.getConnectionFlow()

                                   + ": return to HTTP/1.1 pool");

            connection.closeOrReturnToCache(eof == null ? headers : null);

        }

    }

    HttpHeaders responseHeaders() {

        return headers;

    }

    int responseCode() {

        return responseCode;

    }

// ================ Support for plugging into Http1Receiver   =================

// ============================================================================

    // Callback: Error receiver: Consumer of Throwable.

    void onReadError(Throwable t) {

        Log.logError(t);

        Receiver<?> receiver = receiver(readProgress);

        if (t instanceof EOFException) {

            debug.log(Level.DEBUG, "onReadError: received EOF");

            eof = (EOFException) t;

        }

        CompletableFuture<?> cf = receiver == null ? null : receiver.completion();

        debug.log(Level.DEBUG, () -> "onReadError: cf is "

                + (cf == null  ? "null"

                : (cf.isDone() ? "already completed"

                               : "not yet completed")));

        if (cf != null && !cf.isDone()) cf.completeExceptionally(t);

        else { debug.log(Level.DEBUG, "onReadError", t); }

        debug.log(Level.DEBUG, () -> "closing connection: cause is " + t);

        connection.close();

    }

    // ========================================================================

    private State advance(State previous) {

        assert readProgress == previous;

        switch(previous) {

            case READING_HEADERS:

                asyncReceiver.unsubscribe(headersReader);

                return readProgress = State.READING_BODY;

            case READING_BODY:

                asyncReceiver.unsubscribe(bodyReader);

                return readProgress = State.DONE;

            default:

                throw new InternalError("can't advance from " + previous);

        }

    }

    Receiver<?> receiver(State state) {

        switch(state) {

            case READING_HEADERS: return headersReader;

            case READING_BODY: return bodyReader;

            default: return null;

        }

    }

    static abstract class Receiver<T>

            implements Http1AsyncReceiver.Http1AsyncDelegate {

        abstract void start(T parser);

        abstract CompletableFuture<State> completion();

        // accepts a buffer from upstream.

        // this should be implemented as a simple call to

        // accept(ref, parser, cf)

        public abstract boolean tryAsyncReceive(ByteBuffer buffer);

        public abstract void onReadError(Throwable t);

        // handle a byte buffer received from upstream.

        // this method should set the value of Http1Response.buffer

        // to ref.get() before beginning parsing.

        abstract void handle(ByteBuffer buf, T parser,

                             CompletableFuture<State> cf);

        // resets this objects state so that it can be reused later on

        // typically puts the reference to parser and completion to null

        abstract void reset();

        // accepts a byte buffer received from upstream

        // returns true if the buffer is fully parsed and more data can

        // be accepted, false otherwise.

        final boolean accept(ByteBuffer buf, T parser,

                CompletableFuture<State> cf) {

            if (cf == null || parser == null || cf.isDone()) return false;

            handle(buf, parser, cf);

            return !cf.isDone();

        }

        public abstract void onSubscribe(AbstractSubscription s);

        public abstract AbstractSubscription subscription();

    }

    // Invoked with each new ByteBuffer when reading headers...

    final class HeadersReader extends Receiver<Http1HeaderParser> {

        final Consumer<State> onComplete;

        volatile Http1HeaderParser parser;

        volatile CompletableFuture<State> cf;

        volatile long count; // bytes parsed (for debug)

        volatile AbstractSubscription subscription;

        HeadersReader(Consumer<State> onComplete) {

            this.onComplete = onComplete;

        }

        @Override

        public AbstractSubscription subscription() {

            return subscription;

        }

        @Override

        public void onSubscribe(AbstractSubscription s) {

            this.subscription = s;

            s.request(1);

        }

        @Override

        void reset() {

            cf = null;

            parser = null;

            count = 0;

            subscription = null;

        }

        // Revisit: do we need to support restarting?

        @Override

        final void start(Http1HeaderParser hp) {

            count = 0;

            cf = new MinimalFuture<>();

            parser = hp;

        }

        @Override

        CompletableFuture<State> completion() {

            return cf;

        }

        @Override

        public final boolean tryAsyncReceive(ByteBuffer ref) {

            boolean hasDemand = subscription.demand().tryDecrement();

            assert hasDemand;

            boolean needsMore = accept(ref, parser, cf);

            if (needsMore) subscription.request(1);

            return needsMore;

        }

        @Override

        public final void onReadError(Throwable t) {

            Http1Response.this.onReadError(t);

        }

        @Override

        final void handle(ByteBuffer b,

                          Http1HeaderParser parser,

                          CompletableFuture<State> cf) {

            assert cf != null : "parsing not started";

            assert parser != null : "no parser";

            try {

                count += b.remaining();

                debug.log(Level.DEBUG, () -> "Sending " + b.remaining()

                        + "/" + b.capacity() + " bytes to header parser");

                if (parser.parse(b)) {

                    count -= b.remaining();

                    debug.log(Level.DEBUG, () ->

                            "Parsing headers completed. bytes=" + count);

                    onComplete.accept(State.READING_HEADERS);

                    cf.complete(State.READING_HEADERS);

                }

            } catch (Throwable t) {

                debug.log(Level.DEBUG,

                        () -> "Header parser failed to handle buffer: " + t);

                cf.completeExceptionally(t);

            }

        }

    }

    // Invoked with each new ByteBuffer when reading bodies...

    final class BodyReader extends Receiver<BodyParser> {

        final Consumer<State> onComplete;

        volatile BodyParser parser;

        volatile CompletableFuture<State> cf;

        volatile AbstractSubscription subscription;

        BodyReader(Consumer<State> onComplete) {

            this.onComplete = onComplete;

        }

        @Override

        void reset() {

            parser = null;

            cf = null;

            subscription = null;

        }

        // Revisit: do we need to support restarting?

        @Override

        final void start(BodyParser parser) {

            cf = new MinimalFuture<>();

            this.parser = parser;

        }

        @Override

        CompletableFuture<State> completion() {

            return cf;

        }

        @Override

        public final boolean tryAsyncReceive(ByteBuffer b) {

            return accept(b, parser, cf);

        }

        @Override

        public final void onReadError(Throwable t) {

            Http1Response.this.onReadError(t);

        }

        @Override

        public AbstractSubscription subscription() {

            return subscription;

        }

        @Override

        public void onSubscribe(AbstractSubscription s) {

            this.subscription = s;

            parser.onSubscribe(s);

        }

        @Override

        final void handle(ByteBuffer b,

                          BodyParser parser,

                          CompletableFuture<State> cf) {

            assert cf != null : "parsing not started";

            assert parser != null : "no parser";

            try {

                debug.log(Level.DEBUG, () -> "Sending " + b.remaining()

                        + "/" + b.capacity() + " bytes to body parser");

                parser.accept(b);

            } catch (Throwable t) {

                debug.log(Level.DEBUG,

                        () -> "Body parser failed to handle buffer: " + t);

                if (!cf.isDone()) {

                    cf.completeExceptionally(t);

                }

            }

        }