/*

 * 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.lang.System.Logger.Level;

import java.nio.ByteBuffer;

import java.util.List;

import java.util.concurrent.CompletableFuture;

import java.util.concurrent.CompletionStage;

import java.util.concurrent.Executor;

import java.util.concurrent.Flow;

import java.util.concurrent.atomic.AtomicBoolean;

import java.util.concurrent.atomic.AtomicLong;

import java.util.function.Consumer;

import java.util.function.Function;

import java.net.http.HttpHeaders;

import java.net.http.HttpResponse;

import jdk.internal.net.http.ResponseContent.BodyParser;

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

import static java.net.http.HttpClient.Version.HTTP_1_1;

import static java.net.http.HttpResponse.BodySubscribers.discarding;

import static java.net.HttpURLConnection.HTTP_NOT_MODIFIED;

/**

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

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

    final static AtomicLong responseCount = new AtomicLong();

    final long id = responseCount.incrementAndGet();

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

    }

    String dbgTag;

    private String dbgString() {

        String dbg = dbgTag;

        if (dbg == null) {

            String cdbg = connection.dbgTag;

            if (cdbg != null) {

                dbgTag = dbg = "Http1Response(id=" + id + ", " + cdbg + ")";

            } else {

                dbg = "Http1Response(id=" + id + ")";

            }

        }

        return dbg;

    }

    // The ClientRefCountTracker is used to track the state

    // of a pending operation. Altough there usually is a single

    // point where the operation starts, it may terminate at

    // different places.

    private final class ClientRefCountTracker {

        final HttpClientImpl client = connection.client();

        // state & 0x01 != 0 => acquire called

        // state & 0x02 != 0 => tryRelease called

        byte state;

        public synchronized void acquire() {

            if (state == 0) {

                // increment the reference count on the HttpClientImpl

                // to prevent the SelectorManager thread from exiting

                // until our operation is complete.

                if (debug.on())

                    debug.log("Operation started: incrementing ref count for %s", client);

                client.reference();

                state = 0x01;

            } else {

                if (debug.on())

                    debug.log("Operation ref count for %s is already %s",

                              client, ((state & 0x2) == 0x2) ? "released." : "incremented!" );

                assert (state & 0x01) == 0 : "reference count already incremented";

            }

        }

        public synchronized void tryRelease() {

            if (state == 0x01) {

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

                if (debug.on())

                    debug.log("Operation finished: decrementing ref count for %s", client);

                client.unreference();

            } else if (state == 0) {

                if (debug.on())

                    debug.log("Operation finished: releasing ref count for %s", client);

            } else if ((state & 0x02) == 0x02) {

                if (debug.on())

                    debug.log("ref count for %s already released", client);

            }

            state |= 0x02;

        }

    }

    public CompletableFuture<Response> readHeadersAsync(Executor executor) {

        if (debug.on())

            debug.log("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;

                if (debug.on())

                    debug.log("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,

                                        connection,

                                        responseCode,

                                        HTTP_1_1);

                if (Log.headers()) {

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

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

                    Log.logHeaders(sb.toString());

                }

                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") || responseCode == HTTP_NOT_MODIFIED) {

            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(discarding(), true, executor);

        }

    }

    static final Flow.Subscription NOP = new Flow.Subscription() {

        @Override

        public void request(long n) { }

        public void cancel() { }

    };

    /**

     * The Http1AsyncReceiver ensures that all calls to

     * the subscriber, including onSubscribe, occur sequentially.

     * There could however be some race conditions that could happen

     * in case of unexpected errors thrown at unexpected places, which

     * may cause onError to be called multiple times.

     * The Http1BodySubscriber will ensure that the user subscriber

     * is actually completed only once - and only after it is

     * subscribed.

     * @param <U> The type of response.

     */

    final static class Http1BodySubscriber<U> implements HttpResponse.BodySubscriber<U> {

        final HttpResponse.BodySubscriber<U> userSubscriber;

        final AtomicBoolean completed = new AtomicBoolean();

        volatile Throwable withError;

        volatile boolean subscribed;

        Http1BodySubscriber(HttpResponse.BodySubscriber<U> userSubscriber) {

            this.userSubscriber = userSubscriber;

        }

        // propagate the error to the user subscriber, even if not

        // subscribed yet.

        private void propagateError(Throwable t) {

            assert t != null;

            try {

                // if unsubscribed at this point, it will not

                // get subscribed later - so do it now and

                // propagate the error

                if (subscribed == false) {

                    subscribed = true;

                    userSubscriber.onSubscribe(NOP);

                }

            } finally  {

                // if onError throws then there is nothing to do

                // here: let the caller deal with it by logging

                // and closing the connection.

                userSubscriber.onError(t);

            }

        }

        // complete the subscriber, either normally or exceptionally

        // ensure that the subscriber is completed only once.

        private void complete(Throwable t) {

            if (completed.compareAndSet(false, true)) {

                t  = withError = Utils.getCompletionCause(t);

                if (t == null) {

                    assert subscribed;

                    try {

                        userSubscriber.onComplete();

                    } catch (Throwable x) {

                        propagateError(t = withError = Utils.getCompletionCause(x));

                        // rethrow and let the caller deal with it.

                        // (i.e: log and close the connection)

                        // arguably we could decide to not throw and let the

                        // connection be reused since we should have received and

                        // parsed all the bytes when we reach here.

                        throw x;

                    }

                } else {

                    propagateError(t);

                }

            }

        }

        @Override

        public CompletionStage<U> getBody() {

            return userSubscriber.getBody();

        }

        @Override

        public void onSubscribe(Flow.Subscription subscription) {

            if (!subscribed) {

                subscribed = true;

                userSubscriber.onSubscribe(subscription);

            } else {

                // could be already subscribed and completed

                // if an unexpected error occurred before the actual

                // subscription - though that's not supposed

                // happen.

                assert completed.get();

            }

        }

        @Override

        public void onNext(List<ByteBuffer> item) {

            assert !completed.get();

            userSubscriber.onNext(item);

        }

        @Override

        public void onError(Throwable throwable) {

            complete(throwable);

        }

        @Override

        public void onComplete() {

            complete(null);

        }

    }

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

                                         boolean return2Cache,

                                         Executor executor) {

        this.return2Cache = return2Cache;

        final Http1BodySubscriber<U> subscriber = new Http1BodySubscriber<>(p);

        final CompletableFuture<U> cf = new MinimalFuture<>();

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

        ClientRefCountTracker refCountTracker = new ClientRefCountTracker();

        // We need to keep hold on the client facade until the

        // tracker has been incremented.

        connection.client().reference();

        executor.execute(() -> {

            try {

                content = new ResponseContent(

                        connection, clen, headers, subscriber,

                        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.

                refCountTracker.acquire();

                bodyReader.start(content.getBodyParser(

                    (t) -> {

                        try {

                            if (t != null) {

                                try {

                                    subscriber.onError(t);

                                } finally {

                                    cf.completeExceptionally(t);

                                }

                            }

                        } finally {

                            bodyReader.onComplete(t);

                            if (t != null) {

                                connection.close();

                            }

                        }

                    }));

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

                            if (debug.on()) debug.log("Finished reading body: " + s);

                            assert s == State.READING_BODY;

                        }

                        if (t != null) {

                            subscriber.onError(t);

                            cf.completeExceptionally(t);

                        }

                    } catch (Throwable x) {

                        // not supposed to happen

                        asyncReceiver.onReadError(x);

                    } finally {

                        // we're done: release the ref count for

                        // the current operation.

                        refCountTracker.tryRelease();

                    }

                });

                connection.addTrailingOperation(trailingOp);

            } catch (Throwable t) {

               if (debug.on()) debug.log("Failed reading body: " + t);

                try {

                    subscriber.onError(t);

                    cf.completeExceptionally(t);

                } finally {

                    asyncReceiver.onReadError(t);

                }

            } finally {

                connection.client().unreference();

            }

        });

        try {

            p.getBody().whenComplete((U u, Throwable t) -> {

                if (t == null)

                    cf.complete(u);

                else

                    cf.completeExceptionally(t);

            });

        } catch (Throwable t) {

            cf.completeExceptionally(t);

            asyncReceiver.setRetryOnError(false);

            asyncReceiver.onReadError(t);

        }

        return cf.whenComplete((s,t) -> {

            if (t != null) {

                // If an exception occurred, release the

                // ref count for the current operation, as

                // it may never be triggered otherwise

                // (BodySubscriber ofInputStream)

                // If there was no exception then the

                // ref count will be/have been released when

                // the last byte of the response is/was received

                refCountTracker.tryRelease();

            }

        });

    }

    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.

            if (debug.on())

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