/*

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

import java.nio.ByteBuffer;

import java.util.List;

import java.util.Objects;

import java.util.concurrent.Flow;

import java.util.concurrent.atomic.AtomicLong;

import java.util.concurrent.atomic.AtomicReference;

import java.nio.channels.SelectableChannel;

import java.nio.channels.SelectionKey;

import java.nio.channels.SocketChannel;

import java.util.ArrayList;

import java.util.function.Consumer;

import java.util.function.Supplier;

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

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

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

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

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

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

import jdk.internal.net.http.common.SequentialScheduler.DeferredCompleter;

import jdk.internal.net.http.common.SequentialScheduler.RestartableTask;

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

/**

 * A SocketTube is a terminal tube plugged directly into the socket.

 * The read subscriber should call {@code subscribe} on the SocketTube before

 * the SocketTube is subscribed to the write publisher.

 */

final class SocketTube implements FlowTube {

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

    static final AtomicLong IDS = new AtomicLong();

    private final HttpClientImpl client;

    private final SocketChannel channel;

    private final SliceBufferSource sliceBuffersSource;

    private final Object lock = new Object();

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

    private final InternalReadPublisher readPublisher;

    private final InternalWriteSubscriber writeSubscriber;

    private final long id = IDS.incrementAndGet();

    public SocketTube(HttpClientImpl client, SocketChannel channel,

                      Supplier<ByteBuffer> buffersFactory) {

        this.client = client;

        this.channel = channel;

        this.sliceBuffersSource = new SliceBufferSource(buffersFactory);

        this.readPublisher = new InternalReadPublisher();

        this.writeSubscriber = new InternalWriteSubscriber();

    }

    /**

     * Returns {@code true} if this flow is finished.

     * This happens when this flow internal read subscription is completed,

     * either normally (EOF reading) or exceptionally  (EOF writing, or

     * underlying socket closed, or some exception occurred while reading or

     * writing to the socket).

     *

     * @return {@code true} if this flow is finished.

     */

    public boolean isFinished() {

        InternalReadPublisher.InternalReadSubscription subscription =

                readPublisher.subscriptionImpl;

        return subscription != null && subscription.completed

                || subscription == null && errorRef.get() != null;

    }

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

    //                       Flow.Publisher                                  //

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

    /**

     * {@inheritDoc }

     * @apiNote This method should be called first. In particular, the caller

     *          must ensure that this method must be called by the read

     *          subscriber before the write publisher can call {@code onSubscribe}.

     *          Failure to adhere to this contract may result in assertion errors.

     */

    @Override

    public void subscribe(Flow.Subscriber<? super List<ByteBuffer>> s) {

        Objects.requireNonNull(s);

        assert s instanceof TubeSubscriber : "Expected TubeSubscriber, got:" + s;

        readPublisher.subscribe(s);

    }

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

    //                       Flow.Subscriber                                 //

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

    /**

     * {@inheritDoc }

     * @apiNote The caller must ensure that {@code subscribe} is called by

     *          the read subscriber before {@code onSubscribe} is called by

     *          the write publisher.

     *          Failure to adhere to this contract may result in assertion errors.

     */

    @Override

    public void onSubscribe(Flow.Subscription subscription) {

        writeSubscriber.onSubscribe(subscription);

    }

    @Override

    public void onNext(List<ByteBuffer> item) {

        writeSubscriber.onNext(item);

    }

    @Override

    public void onError(Throwable throwable) {

        writeSubscriber.onError(throwable);

    }

    @Override

    public void onComplete() {

        writeSubscriber.onComplete();

    }

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

    //                           Events                                      //

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

    void signalClosed() {

        // Ensures that the subscriber will be terminated and that future

        // subscribers will be notified when the connection is closed.

        if (Log.channel()) {

            Log.logChannel("Connection close signalled: connection closed locally ({0})",

                    channelDescr());

        }

        readPublisher.subscriptionImpl.signalError(

                new IOException("connection closed locally"));

    }

    /**

     * A restartable task used to process tasks in sequence.

     */

    private static class SocketFlowTask implements RestartableTask {

        final Runnable task;

        private final Object monitor = new Object();

        SocketFlowTask(Runnable task) {

            this.task = task;

        }

        @Override

        public final void run(DeferredCompleter taskCompleter) {

            try {

                // non contentious synchronized for visibility.

                synchronized(monitor) {

                    task.run();

                }

            } finally {

                taskCompleter.complete();

            }

        }

    }

    // This is best effort - there's no guarantee that the printed set of values

    // is consistent. It should only be considered as weakly accurate - in

    // particular in what concerns the events states, especially when displaying

    // a read event state from a write event callback and conversely.

    void debugState(String when) {

        if (debug.on()) {

            StringBuilder state = new StringBuilder();

            InternalReadPublisher.InternalReadSubscription sub =

                    readPublisher.subscriptionImpl;

            InternalReadPublisher.ReadEvent readEvent =

                    sub == null ? null : sub.readEvent;

            Demand rdemand = sub == null ? null : sub.demand;

            InternalWriteSubscriber.WriteEvent writeEvent =

                    writeSubscriber.writeEvent;

            Demand wdemand = writeSubscriber.writeDemand;

            int rops = readEvent == null ? 0 : readEvent.interestOps();

            long rd = rdemand == null ? 0 : rdemand.get();

            int wops = writeEvent == null ? 0 : writeEvent.interestOps();

            long wd = wdemand == null ? 0 : wdemand.get();

            state.append(when).append(" Reading: [ops=")

                    .append(rops).append(", demand=").append(rd)

                    .append(", stopped=")

                    .append((sub == null ? false : sub.readScheduler.isStopped()))

                    .append("], Writing: [ops=").append(wops)

                    .append(", demand=").append(wd)

                    .append("]");

            debug.log(state.toString());

        }

    }

    /**

     * A repeatable event that can be paused or resumed by changing its

     * interestOps. When the event is fired, it is first paused before being

     * signaled. It is the responsibility of the code triggered by

     * {@code signalEvent} to resume the event if required.

     */

    private static abstract class SocketFlowEvent extends AsyncEvent {

        final SocketChannel channel;

        final int defaultInterest;

        volatile int interestOps;

        volatile boolean registered;

        SocketFlowEvent(int defaultInterest, SocketChannel channel) {

            super(AsyncEvent.REPEATING);

            this.defaultInterest = defaultInterest;

            this.channel = channel;

        }

        final boolean registered() {return registered;}

        final void resume() {

            interestOps = defaultInterest;

            registered = true;

        }

        final void pause() {interestOps = 0;}

        @Override

        public final SelectableChannel channel() {return channel;}

        @Override

        public final int interestOps() {return interestOps;}

        @Override

        public final void handle() {

            pause();       // pause, then signal

            signalEvent(); // won't be fired again until resumed.

        }

        @Override

        public final void abort(IOException error) {

            debug().log(() -> "abort: " + error);

            pause();              // pause, then signal

            signalError(error);   // should not be resumed after abort (not checked)

        }

        protected abstract void signalEvent();

        protected abstract void signalError(Throwable error);

        abstract Logger debug();

    }

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

    //                              Writing                                  //

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

    // This class makes the assumption that the publisher will call onNext

    // sequentially, and that onNext won't be called if the demand has not been

    // incremented by request(1).

    // It has a 'queue of 1' meaning that it will call request(1) in

    // onSubscribe, and then only after its 'current' buffer list has been

    // fully written and current set to null;

    private final class InternalWriteSubscriber

            implements Flow.Subscriber<List<ByteBuffer>> {

        volatile WriteSubscription subscription;

        volatile List<ByteBuffer> current;

        volatile boolean completed;

        final AsyncTriggerEvent startSubscription =

                new AsyncTriggerEvent(this::signalError, this::startSubscription);

        final WriteEvent writeEvent = new WriteEvent(channel, this);

        final Demand writeDemand = new Demand();

        @Override

        public void onSubscribe(Flow.Subscription subscription) {

            WriteSubscription previous = this.subscription;

            if (debug.on()) debug.log("subscribed for writing");

            try {

                boolean needEvent = current == null;

                if (needEvent) {

                    if (previous != null && previous.upstreamSubscription != subscription) {

                        previous.dropSubscription();

                    }

                }

                this.subscription = new WriteSubscription(subscription);

                if (needEvent) {

                    if (debug.on())

                        debug.log("write: registering startSubscription event");

                    client.registerEvent(startSubscription);

                }

            } catch (Throwable t) {

                signalError(t);

            }

        }

        @Override

        public void onNext(List<ByteBuffer> bufs) {

            assert current == null : dbgString() // this is a queue of 1.

                    + "w.onNext current: " + current;

            assert subscription != null : dbgString()

                    + "w.onNext: subscription is null";

            current = bufs;

            tryFlushCurrent(client.isSelectorThread()); // may be in selector thread

            // For instance in HTTP/2, a received SETTINGS frame might trigger

            // the sending of a SETTINGS frame in turn which might cause

            // onNext to be called from within the same selector thread that the

            // original SETTINGS frames arrived on. If rs is the read-subscriber

            // and ws is the write-subscriber then the following can occur:

            // ReadEvent -> rs.onNext(bytes) -> process server SETTINGS -> write

            // client SETTINGS -> ws.onNext(bytes) -> tryFlushCurrent

            debugState("leaving w.onNext");

        }

        // Don't use a SequentialScheduler here: rely on onNext() being invoked

        // sequentially, and not being invoked if there is no demand, request(1).

        // onNext is usually called from within a user / executor thread.

        // Initial writing will be performed in that thread. If for some reason,

        // not all the data can be written, a writeEvent will be registered, and

        // writing will resume in the selector manager thread when the

        // writeEvent is fired.

        //

        // If this method is invoked in the selector manager thread (because of

        // a writeEvent), then the executor will be used to invoke request(1),

        // ensuring that onNext() won't be invoked from within the selector

        // thread. If not in the selector manager thread, then request(1) is

        // invoked directly.

        void tryFlushCurrent(boolean inSelectorThread) {

            List<ByteBuffer> bufs = current;

            if (bufs == null) return;

            try {

                assert inSelectorThread == client.isSelectorThread() :

                       "should " + (inSelectorThread ? "" : "not ")

                        + " be in the selector thread";

                long remaining = Utils.remaining(bufs);

                if (debug.on()) debug.log("trying to write: %d", remaining);

                long written = writeAvailable(bufs);

                if (debug.on()) debug.log("wrote: %d", written);

                assert written >= 0 : "negative number of bytes written:" + written;

                assert written <= remaining;

                if (remaining - written == 0) {

                    current = null;

                    if (writeDemand.tryDecrement()) {

                        Runnable requestMore = this::requestMore;

                        if (inSelectorThread) {

                            assert client.isSelectorThread();

                            client.theExecutor().execute(requestMore);

                        } else {

                            assert !client.isSelectorThread();

                            requestMore.run();

                        }

                    }

                } else {

                    resumeWriteEvent(inSelectorThread);

                }

            } catch (Throwable t) {

                signalError(t);

            }

        }

        // Kick off the initial request:1 that will start the writing side.

        // Invoked in the selector manager thread.

        void startSubscription() {

            try {

                if (debug.on()) debug.log("write: starting subscription");

                if (Log.channel()) {

                    Log.logChannel("Start requesting bytes for writing to channel: {0}",

                            channelDescr());

                }

                assert client.isSelectorThread();

                // make sure read registrations are handled before;

                readPublisher.subscriptionImpl.handlePending();

                if (debug.on()) debug.log("write: offloading requestMore");

                // start writing;

                client.theExecutor().execute(this::requestMore);

            } catch(Throwable t) {

                signalError(t);

            }

        }

        void requestMore() {

           WriteSubscription subscription = this.subscription;

           subscription.requestMore();

        }

        @Override

        public void onError(Throwable throwable) {

            signalError(throwable);

        }

        @Override

        public void onComplete() {

            completed = true;

            // no need to pause the write event here: the write event will

            // be paused if there is nothing more to write.

            List<ByteBuffer> bufs = current;

            long remaining = bufs == null ? 0 : Utils.remaining(bufs);

            if (debug.on())

                debug.log( "write completed, %d yet to send", remaining);

            debugState("InternalWriteSubscriber::onComplete");

        }

        void resumeWriteEvent(boolean inSelectorThread) {

            if (debug.on()) debug.log("scheduling write event");

            resumeEvent(writeEvent, this::signalError);

        }

        void signalWritable() {

            if (debug.on()) debug.log("channel is writable");

            tryFlushCurrent(true);

        }

        void signalError(Throwable error) {

            debug.log(() -> "write error: " + error);

            if (Log.channel()) {

                Log.logChannel("Failed to write to channel ({0}: {1})",

                        channelDescr(), error);

            }

            completed = true;

            readPublisher.signalError(error);

            Flow.Subscription subscription = this.subscription;

            if (subscription != null) subscription.cancel();

        }

        // A repeatable WriteEvent which is paused after firing and can

        // be resumed if required - see SocketFlowEvent;

        final class WriteEvent extends SocketFlowEvent {

            final InternalWriteSubscriber sub;

            WriteEvent(SocketChannel channel, InternalWriteSubscriber sub) {

                super(SelectionKey.OP_WRITE, channel);

                this.sub = sub;

            }

            @Override

            protected final void signalEvent() {

                try {

                    client.eventUpdated(this);

                    sub.signalWritable();

                } catch(Throwable t) {

                    sub.signalError(t);

                }

            }

            @Override

            protected void signalError(Throwable error) {

                sub.signalError(error);

            }

            @Override

            Logger debug() { return debug; }

        }

        final class WriteSubscription implements Flow.Subscription {

            final Flow.Subscription upstreamSubscription;

            volatile boolean cancelled;

            WriteSubscription(Flow.Subscription subscription) {

                this.upstreamSubscription = subscription;

            }

            @Override

            public void request(long n) {

                if (cancelled) return;

                upstreamSubscription.request(n);

            }

            @Override

            public void cancel() {

                if (cancelled) return;

                if (debug.on()) debug.log("write: cancel");

                if (Log.channel()) {

                    Log.logChannel("Cancelling write subscription");

                }

                dropSubscription();

                upstreamSubscription.cancel();

            }

            void dropSubscription() {

                synchronized (InternalWriteSubscriber.this) {

                    cancelled = true;

                    if (debug.on()) debug.log("write: resetting demand to 0");

                    writeDemand.reset();

                }

            }

            void requestMore() {

                try {

                    if (completed || cancelled) return;

                    boolean requestMore;

                    long d;

                    // don't fiddle with demand after cancel.

                    // see dropSubscription.

                    synchronized (InternalWriteSubscriber.this) {

                        if (cancelled) return;

                        d = writeDemand.get();

                        requestMore = writeDemand.increaseIfFulfilled();

                    }

                    if (requestMore) {

                        if (debug.on()) debug.log("write: requesting more...");

                        upstreamSubscription.request(1);

                    } else {

                        if (debug.on())

                            debug.log("write: no need to request more: %d", d);

                    }

                } catch (Throwable t) {

                    if (debug.on())

                        debug.log("write: error while requesting more: " + t);

                    signalError(t);