/*
 * Copyright (c) 2017, 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.lang.System.Logger.Level;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedDeque;
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.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import jdk.internal.net.http.common.Demand;
import jdk.internal.net.http.common.FlowTube.TubeSubscriber;
import jdk.internal.net.http.common.Logger;
import jdk.internal.net.http.common.SequentialScheduler;
import jdk.internal.net.http.common.ConnectionExpiredException;
import jdk.internal.net.http.common.Utils;


/**
 * A helper class that will queue up incoming data until the receiving
 * side is ready to handle it.
 */
class Http1AsyncReceiver {

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

    /**
     * A delegate that can asynchronously receive data from an upstream flow,
     * parse, it, then possibly transform it and either store it (response
     * headers) or possibly pass it to a downstream subscriber (response body).
     * Usually, there will be one Http1AsyncDelegate in charge of receiving
     * and parsing headers, and another one in charge of receiving, parsing,
     * and forwarding body. Each will sequentially subscribe with the
     * Http1AsyncReceiver in turn. There may be additional delegates which
     * subscribe to the Http1AsyncReceiver, mainly for the purpose of handling
     * errors while the connection is busy transmitting the request body and the
     * Http1Exchange::readBody method hasn't been called yet, and response
     * delegates haven't subscribed yet.
     */
    static interface Http1AsyncDelegate {
        /**
         * Receives and handles a byte buffer reference.
         * @param ref A byte buffer reference coming from upstream.
         * @return false, if the byte buffer reference should be kept in the queue.
         *         Usually, this means that either the byte buffer reference
         *         was handled and parsing is finished, or that the receiver
         *         didn't handle the byte reference at all.
         *         There may or may not be any remaining data in the
         *         byte buffer, and the byte buffer reference must not have
         *         been cleared.
         *         true, if the byte buffer reference was fully read and
         *         more data can be received.
         */
        public boolean tryAsyncReceive(ByteBuffer ref);

        /**
         * Called when an exception is raised.
         * @param ex The raised Throwable.
         */
        public void onReadError(Throwable ex);

        /**
         * Must be called before any other method on the delegate.
         * The subscription can be either used directly by the delegate
         * to request more data (e.g. if the delegate is a header parser),
         * or can be forwarded to a downstream subscriber (if the delegate
         * is a body parser that wraps a response BodySubscriber).
         * In all cases, it is the responsibility of the delegate to ensure
         * that request(n) and demand.tryDecrement() are called appropriately.
         * No data will be sent to {@code tryAsyncReceive} unless
         * the subscription has some demand.
         *
         * @param s A subscription that allows the delegate to control the
         *          data flow.
         */
        public void onSubscribe(AbstractSubscription s);

        /**
         * Returns the subscription that was passed to {@code onSubscribe}
         * @return the subscription that was passed to {@code onSubscribe}..
         */
        public AbstractSubscription subscription();

        /**
         * Called to make sure resources are released when the
         * when the Http1AsyncReceiver is stopped.
         * @param error The Http1AsyncReceiver pending error ref,
         *              if any.
         */
        public void close(Throwable error);

    }

    /**
     * A simple subclass of AbstractSubscription that ensures the
     * SequentialScheduler will be run when request() is called and demand
     * becomes positive again.
     */
    private static final class Http1AsyncDelegateSubscription
            extends AbstractSubscription
    {
        private final Runnable onCancel;
        private final Consumer<Throwable> onError;
        private final SequentialScheduler scheduler;
        private volatile boolean cancelled;
        Http1AsyncDelegateSubscription(SequentialScheduler scheduler,
                                       Runnable onCancel,
                                       Consumer<Throwable> onError) {
            this.scheduler = scheduler;
            this.onCancel = onCancel;
            this.onError = onError;
        }
        @Override
        public void request(long n) {
            if (cancelled) return;
            try {
                final Demand demand = demand();
                if (demand.increase(n)) {
                    scheduler.runOrSchedule();
                }
            } catch (IllegalArgumentException x) {
                cancelled = true;
                onError.accept(x);
            }
        }
        @Override
        public void cancel() {
            cancelled = true;
            onCancel.run();
        }
    }

    private final ConcurrentLinkedDeque<ByteBuffer> queue
            = new ConcurrentLinkedDeque<>();
    private final SequentialScheduler scheduler =
            SequentialScheduler.synchronizedScheduler(this::flush);
    private final Executor executor;
    private final Http1TubeSubscriber subscriber = new Http1TubeSubscriber();
    private final AtomicReference<Http1AsyncDelegate> pendingDelegateRef;
    private final AtomicLong received = new AtomicLong();
    final AtomicBoolean canRequestMore = new AtomicBoolean();

    private volatile Throwable error;
    private volatile Http1AsyncDelegate delegate;
    // This reference is only used to prevent early GC of the exchange.
    private volatile Http1Exchange<?>  owner;
    // Only used for checking whether we run on the selector manager thread.
    private final HttpClientImpl client;
    private boolean retry;
    private volatile boolean stopRequested;

    public Http1AsyncReceiver(Executor executor, Http1Exchange<?> owner) {
        this.pendingDelegateRef = new AtomicReference<>();
        this.executor = executor;
        this.owner = owner;
        this.client = owner.client;
    }

    // This is the main loop called by the SequentialScheduler.
    // It attempts to empty the queue until the scheduler is stopped,
    // or the delegate is unregistered, or the delegate is unable to
    // process the data (because it's not ready or already done), which
    // it signals by returning 'true';
    private void flush() {
        ByteBuffer buf;
        try {
            assert !client.isSelectorThread() :
                    "Http1AsyncReceiver::flush should not run in the selector: "
                    + Thread.currentThread().getName();

            // First check whether we have a pending delegate that has
            // just subscribed, and if so, create a Subscription for it
            // and call onSubscribe.
            handlePendingDelegate();

            // Then start emptying the queue, if possible.
            while ((buf = queue.peek()) != null && !stopRequested) {
                Http1AsyncDelegate delegate = this.delegate;
                if (debug.on())
                    debug.log("Got %s bytes for delegate %s",
                              buf.remaining(), delegate);
                if (!hasDemand(delegate)) {
                    // The scheduler will be invoked again later when the demand
                    // becomes positive.
                    return;
                }

                assert delegate != null;
                if (debug.on())
                    debug.log("Forwarding %s bytes to delegate %s",
                              buf.remaining(), delegate);
                // The delegate has demand: feed it the next buffer.
                if (!delegate.tryAsyncReceive(buf)) {
                    final long remaining = buf.remaining();
                    if (debug.on()) debug.log(() -> {
                        // If the scheduler is stopped, the queue may already
                        // be empty and the reference may already be released.
                        String remstr = scheduler.isStopped() ? "" :
                                " remaining in ref: "
                                + remaining;
                        remstr +=  remstr
                                + " total remaining: " + remaining();
                        return "Delegate done: " + remaining;
                    });
                    canRequestMore.set(false);
                    // The last buffer parsed may have remaining unparsed bytes.
                    // Don't take it out of the queue.
                    return; // done.
                }

                // removed parsed buffer from queue, and continue with next
                // if available
                ByteBuffer parsed = queue.remove();
                canRequestMore.set(queue.isEmpty() && !stopRequested);
                assert parsed == buf;
            }

            // queue is empty: let's see if we should request more
            checkRequestMore();

        } catch (Throwable t) {
            Throwable x = error;
            if (x == null) error = t; // will be handled in the finally block
            if (debug.on()) debug.log("Unexpected error caught in flush()", t);
        } finally {
            // Handles any pending error.
            // The most recently subscribed delegate will get the error.
            checkForErrors();
        }
    }

    /**
     * Must be called from within the scheduler main loop.
     * Handles any pending errors by calling delegate.onReadError().
     * If the error can be forwarded to the delegate, stops the scheduler.
     */
    private void checkForErrors() {
        // Handles any pending error.
        // The most recently subscribed delegate will get the error.
        // If the delegate is null, the error will be handled by the next
        // delegate that subscribes.
        // If the queue is not empty, wait until it it is empty before
        // handling the error.
        Http1AsyncDelegate delegate = pendingDelegateRef.get();
        if (delegate == null) delegate = this.delegate;
        Throwable x = error;
        if (delegate != null && x != null && (stopRequested || queue.isEmpty())) {
            // forward error only after emptying the queue.
            final Object captured = delegate;
            if (debug.on())
                debug.log(() -> "flushing " + x + "\n\t delegate: " + captured
                          + "\t\t queue.isEmpty: " + queue.isEmpty());
            scheduler.stop();
            delegate.onReadError(x);
            if (stopRequested) {
                // This is the special case where the subscriber
                // has requested an illegal number of items.
                // In this case, the error doesn't come from
                // upstream, but from downstream, and we need to
                // close the upstream connection.
                Http1Exchange<?> exchg = owner;
                stop();
                if (exchg != null) exchg.connection().close();
            }
        }
    }

    /**
     * Must be called from within the scheduler main loop.
     * Figure out whether more data should be requested from the
     * Http1TubeSubscriber.
     */
    private void checkRequestMore() {
        Http1AsyncDelegate delegate = this.delegate;
        boolean more = this.canRequestMore.get();
        boolean hasDemand = hasDemand(delegate);
        if (debug.on())
            debug.log("checkRequestMore: " + "canRequestMore=" + more
                      + ", hasDemand=" + hasDemand
                      + (delegate == null ? ", delegate=null" : ""));
        if (hasDemand) {
            subscriber.requestMore();
        }
    }

    /**
     * Must be called from within the scheduler main loop.
     * Return true if the delegate is not null and has some demand.
     * @param delegate The Http1AsyncDelegate delegate
     * @return true if the delegate is not null and has some demand
     */
    private boolean hasDemand(Http1AsyncDelegate delegate) {
        if (delegate == null) return false;
        AbstractSubscription subscription = delegate.subscription();
        long demand = subscription.demand().get();
        if (debug.on())
            debug.log("downstream subscription demand is %s", demand);
        return demand > 0;
    }

    /**
     * Must be called from within the scheduler main loop.
     * Handles pending delegate subscription.
     * Return true if there was some pending delegate subscription and a new
     * delegate was subscribed, false otherwise.
     *
     * @return true if there was some pending delegate subscription and a new
     *         delegate was subscribed, false otherwise.
     */
    private boolean handlePendingDelegate() {
        Http1AsyncDelegate pending = pendingDelegateRef.get();
        if (pending != null && pendingDelegateRef.compareAndSet(pending, null)) {
            Http1AsyncDelegate delegate = this.delegate;
            if (delegate != null) unsubscribe(delegate);
            Consumer<Throwable> onSubscriptionError = (x) -> {
                setRetryOnError(false);
                stopRequested = true;
                onReadError(x);
            };
            Runnable cancel = () -> {
                if (debug.on())
                    debug.log("Downstream subscription cancelled by %s", pending);
                // The connection should be closed, as some data may
                // be left over in the stream.
                try {
                    setRetryOnError(false);
                    pending.close(null);
                    onReadError(new IOException("subscription cancelled"));
                    unsubscribe(pending);
                } finally {
                    Http1Exchange<?> exchg = owner;
                    stop();
                    if (exchg != null) exchg.connection().close();
                }
            };
            // The subscription created by a delegate is only loosely
            // coupled with the upstream subscription. This is partly because
            // the header/body parser work with a flow of ByteBuffer, whereas
            // we have a flow List<ByteBuffer> upstream.
            Http1AsyncDelegateSubscription subscription =
                    new Http1AsyncDelegateSubscription(scheduler, cancel, onSubscriptionError);
            pending.onSubscribe(subscription);
            this.delegate = delegate = pending;
            final Object captured = delegate;
            if (debug.on())
                debug.log("delegate is now " + captured
                          + ", demand=" + subscription.demand().get()
                          + ", canRequestMore=" + canRequestMore.get()
                          + ", queue.isEmpty=" + queue.isEmpty());
            return true;
        }
        return false;
    }

    synchronized void setRetryOnError(boolean retry) {
        this.retry = retry;
    }

    void clear() {
        if (debug.on()) debug.log("cleared");
        this.pendingDelegateRef.set(null);
        this.delegate = null;
        this.owner = null;
    }

    void subscribe(Http1AsyncDelegate delegate) {
        synchronized(this) {
            pendingDelegateRef.set(delegate);
        }
        if (queue.isEmpty()) {
            canRequestMore.set(true);
        }
        if (debug.on())
            debug.log("Subscribed pending " + delegate + " queue.isEmpty: "
                      + queue.isEmpty());
        // Everything may have been received already. Make sure
        // we parse it.
        if (client.isSelectorThread()) {
            scheduler.runOrSchedule(executor);
        } else {
            scheduler.runOrSchedule();
        }
    }

    // Used for debugging only!
    long remaining() {
        return Utils.remaining(queue.toArray(Utils.EMPTY_BB_ARRAY));
    }

    void unsubscribe(Http1AsyncDelegate delegate) {
        synchronized(this) {
            if (this.delegate == delegate) {
                if (debug.on()) debug.log("Unsubscribed %s", delegate);
                this.delegate = null;
            }
        }
    }

    // Callback: Consumer of ByteBuffer
    private void asyncReceive(ByteBuffer buf) {
        if (debug.on())
            debug.log("Putting %s bytes into the queue", buf.remaining());
        received.addAndGet(buf.remaining());
        queue.offer(buf);

        // This callback is called from within the selector thread.
        // Use an executor here to avoid doing the heavy lifting in the
        // selector.
        scheduler.runOrSchedule(executor);
    }

    // Callback: Consumer of Throwable
    void onReadError(Throwable ex) {
        Http1AsyncDelegate delegate;
        Throwable recorded;
        if (debug.on()) debug.log("onError: %s", (Object) ex);
        synchronized (this) {
            delegate = this.delegate;
            recorded = error;
            if (recorded == null) {
                // retry is set to true by HttpExchange when the connection is
                // already connected, which means it's been retrieved from
                // the pool.
                if (retry && (ex instanceof IOException)) {
                    // could be either EOFException, or
                    // IOException("connection reset by peer), or
                    // SSLHandshakeException resulting from the server having
                    // closed the SSL session.
                    if (received.get() == 0) {
                        // If we receive such an exception before having
                        // received any byte, then in this case, we will
                        // throw ConnectionExpiredException
                        // to try & force a retry of the request.
                        retry = false;
                        ex = new ConnectionExpiredException(
                                "subscription is finished", ex);
                    }
                }
                error = ex;
            }
            final Throwable t = (recorded == null ? ex : recorded);
            if (debug.on())
                debug.log("recorded " + t + "\n\t delegate: " + delegate
                          + "\t\t queue.isEmpty: " + queue.isEmpty(), ex);
        }
        if (queue.isEmpty() || pendingDelegateRef.get() != null || stopRequested) {
            // This callback is called from within the selector thread.
            // Use an executor here to avoid doing the heavy lifting in the
            // selector.
            scheduler.runOrSchedule(executor);
        }
    }

    void stop() {
        if (debug.on()) debug.log("stopping");
        scheduler.stop();
        // make sure ref count is handled properly by
        // closing the delegate.
        Http1AsyncDelegate previous = delegate;
        if (previous != null) previous.close(error);
        delegate = null;
        owner  = null;
    }

    /**
     * Returns the TubeSubscriber for reading from the connection flow.
     * @return the TubeSubscriber for reading from the connection flow.
     */
    TubeSubscriber subscriber() {
        return subscriber;
    }

    /**
     * A simple tube subscriber for reading from the connection flow.
     */
    final class Http1TubeSubscriber implements TubeSubscriber {
        volatile Flow.Subscription subscription;
        volatile boolean completed;
        volatile boolean dropped;

        public void onSubscribe(Flow.Subscription subscription) {
            // supports being called multiple time.
            // doesn't cancel the previous subscription, since that is
            // most probably the same as the new subscription.
            assert this.subscription == null || dropped == false;
            this.subscription = subscription;
            dropped = false;
            canRequestMore.set(true);
            if (delegate != null) {
                scheduler.runOrSchedule(executor);
            }
        }

        void requestMore() {
            Flow.Subscription s = subscription;
            if (s == null) return;
            if (canRequestMore.compareAndSet(true, false)) {
                if (!completed && !dropped) {
                    if (debug.on())
                        debug.log("Http1TubeSubscriber: requesting one more from upstream");
                    s.request(1);
                    return;
                }
            }
            if (debug.on())
                debug.log("Http1TubeSubscriber: no need to request more");
        }

        @Override
        public void onNext(List<ByteBuffer> item) {
            canRequestMore.set(item.isEmpty());
            for (ByteBuffer buffer : item) {
                asyncReceive(buffer);
            }
        }

        @Override
        public void onError(Throwable throwable) {
            onReadError(throwable);
            completed = true;
        }

        @Override
        public void onComplete() {
            onReadError(new EOFException("EOF reached while reading"));
            completed = true;
        }

        public void dropSubscription() {
            if (debug.on()) debug.log("Http1TubeSubscriber: dropSubscription");
            // we could probably set subscription to null here...
            // then we might not need the 'dropped' boolean?
            dropped = true;
        }

    }

    // Drains the content of the queue into a single ByteBuffer.
    // The scheduler must be permanently stopped before calling drain().
    ByteBuffer drain(ByteBuffer initial) {
        // Revisit: need to clean that up.
        //
        ByteBuffer b = initial = (initial == null ? Utils.EMPTY_BYTEBUFFER : initial);
        assert scheduler.isStopped();

        if (queue.isEmpty()) return b;

        // sanity check: we shouldn't have queued the same
        // buffer twice.
        ByteBuffer[] qbb = queue.toArray(new ByteBuffer[queue.size()]);

// the assertion looks suspicious, more investigation needed
//
//        assert java.util.stream.Stream.of(qbb)
//                .collect(Collectors.toSet())
//                .size() == qbb.length : debugQBB(qbb);

        // compute the number of bytes in the queue, the number of bytes
        // in the initial buffer
        // TODO: will need revisiting - as it is not guaranteed that all
        // data will fit in single BB!
        int size = Utils.remaining(qbb, Integer.MAX_VALUE);
        int remaining = b.remaining();
        int free = b.capacity() - b.position() - remaining;
        if (debug.on())
            debug.log("Flushing %s bytes from queue into initial buffer "
                      + "(remaining=%s, free=%s)", size, remaining, free);

        // check whether the initial buffer has enough space
        if (size > free) {
            if (debug.on())
                debug.log("Allocating new buffer for initial: %s", (size + remaining));
            // allocates a new buffer and copy initial to it
            b = ByteBuffer.allocate(size + remaining);
            Utils.copy(initial, b);
            assert b.position() == remaining;
            b.flip();
            assert b.position() == 0;
            assert b.limit() == remaining;
            assert b.remaining() == remaining;
        }

        // store position and limit
        int pos = b.position();
        int limit = b.limit();
        assert limit - pos == remaining;
        assert b.capacity() >= remaining + size
                : "capacity: " + b.capacity()
                + ", remaining: " + b.remaining()
                + ", size: " + size;

        // prepare to copy the content of the queue
        b.position(limit);
        b.limit(pos + remaining + size);
        assert b.remaining() >= size :
                "remaining: " + b.remaining() + ", size: " + size;

        // copy the content of the queue
        int count = 0;
        for (int i=0; i<qbb.length; i++) {
            ByteBuffer b2 = qbb[i];
            int r = b2.remaining();
            assert b.remaining() >= r : "need at least " + r + " only "
                    + b.remaining() + " available";
            int copied = Utils.copy(b2, b);
            assert copied == r : "copied="+copied+" available="+r;
            assert b2.remaining() == 0;
            count += copied;
        }
        assert count == size;
        assert b.position() == pos + remaining + size :
                "b.position="+b.position()+" != "+pos+"+"+remaining+"+"+size;

        // reset limit and position
        b.limit(limit+size);
        b.position(pos);

        // we can clear the refs
        queue.clear();
        final ByteBuffer bb = b;
        if (debug.on())
            debug.log("Initial buffer now has " + bb.remaining()
                       + " pos=" + bb.position() + " limit=" + bb.limit());

        return b;
    }

    private String debugQBB(ByteBuffer[] qbb) {
        StringBuilder msg = new StringBuilder();
        List<ByteBuffer> lbb = Arrays.asList(qbb);
        Set<ByteBuffer> sbb = new HashSet<>(Arrays.asList(qbb));

        int uniquebb = sbb.size();
        msg.append("qbb: ").append(lbb.size())
           .append(" (unique: ").append(uniquebb).append("), ")
           .append("duplicates: ");
        String sep = "";
        for (ByteBuffer b : lbb) {
            if (!sbb.remove(b)) {
                msg.append(sep)
                   .append(String.valueOf(b))
                   .append("[remaining=")
                   .append(b.remaining())
                   .append(", position=")
                   .append(b.position())
                   .append(", capacity=")
                   .append(b.capacity())
                   .append("]");
                sep = ", ";
            }
        }
        return msg.toString();
    }

    volatile String dbgTag;
    String dbgString() {
        String tag = dbgTag;
        if (tag == null) {
            String flowTag = null;
            Http1Exchange<?> exchg = owner;
            Object flow = (exchg != null)
                    ? exchg.connection().getConnectionFlow()
                    : null;
            flowTag = tag = flow == null ? null: (String.valueOf(flow));
            if (flowTag != null) {
                dbgTag = tag = "Http1AsyncReceiver("+ flowTag + ")";
            } else {
                tag = "Http1AsyncReceiver(?)";
            }
        }
        return tag;
    }
}