/*

 * 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.nio.ByteBuffer;

import java.util.ArrayList;

import java.util.Collections;

import java.util.List;

import java.util.ListIterator;

import java.util.Objects;

import java.util.concurrent.CompletionStage;

import java.util.concurrent.Flow;

import java.util.concurrent.atomic.AtomicBoolean;

import java.net.http.HttpResponse.BodySubscriber;

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

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

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

/**

 * A buffering BodySubscriber. When subscribed, accumulates ( buffers ) a given

 * amount ( in bytes ) of a publisher's data before pushing it to a downstream

 * subscriber.

 */

public class BufferingSubscriber<T> implements BodySubscriber<T>

{

    /** The downstream consumer of the data. */

    private final BodySubscriber<T> downstreamSubscriber;

    /** The amount of data to be accumulate before pushing downstream. */

    private final int bufferSize;

    /** The subscription, created lazily. */

    private volatile Flow.Subscription subscription;

    /** The downstream subscription, created lazily. */

    private volatile DownstreamSubscription downstreamSubscription;

    /** Must be held when accessing the internal buffers. */

    private final Object buffersLock = new Object();

    /** The internal buffers holding the buffered data. */

    private ArrayList<ByteBuffer> internalBuffers;

    /** The actual accumulated remaining bytes in internalBuffers. */

    private int accumulatedBytes;

    /** Holds the Throwable from upstream's onError. */

    private volatile Throwable throwable;

    /** State of the buffering subscriber:

     *  1) [UNSUBSCRIBED] when initially created

     *  2) [ACTIVE] when subscribed and can receive data

     *  3) [ERROR | CANCELLED | COMPLETE] (terminal state)

     */

    static final int UNSUBSCRIBED = 0x01;

    static final int ACTIVE       = 0x02;

    static final int ERROR        = 0x04;

    static final int CANCELLED    = 0x08;

    static final int COMPLETE     = 0x10;

    private volatile int state;

    public BufferingSubscriber(BodySubscriber<T> downstreamSubscriber,

                               int bufferSize) {

        this.downstreamSubscriber = Objects.requireNonNull(downstreamSubscriber);

        this.bufferSize = bufferSize;

        synchronized (buffersLock) {

            internalBuffers = new ArrayList<>();

        }

        state = UNSUBSCRIBED;

    }

    /** Returns the number of bytes remaining in the given buffers. */

    private static final long remaining(List<ByteBuffer> buffers) {

        return buffers.stream().mapToLong(ByteBuffer::remaining).sum();

    }

    /**

     * Tells whether, or not, there is at least a sufficient number of bytes

     * accumulated in the internal buffers. If the subscriber is COMPLETE, and

     * has some buffered data, then there is always enough ( to pass downstream ).

     */

    private final boolean hasEnoughAccumulatedBytes() {

        assert Thread.holdsLock(buffersLock);

        return accumulatedBytes >= bufferSize

                || (state == COMPLETE && accumulatedBytes > 0);

    }

    /**

     * Returns a new, unmodifiable, List<ByteBuffer> containing exactly the

     * amount of data as required before pushing downstream. The amount of data

     * may be less than required ( bufferSize ), in the case where the subscriber

     * is COMPLETE.

     */

    private List<ByteBuffer> fromInternalBuffers() {

        assert Thread.holdsLock(buffersLock);

        int leftToFill = bufferSize;

        int state = this.state;

        assert (state == ACTIVE || state == CANCELLED)

                ? accumulatedBytes >= leftToFill : true;

        List<ByteBuffer> dsts = new ArrayList<>();

        ListIterator<ByteBuffer> itr = internalBuffers.listIterator();

        while (itr.hasNext()) {

            ByteBuffer b = itr.next();

            if (b.remaining() <= leftToFill) {

                itr.remove();

                if (b.position() != 0)

                    b = b.slice();  // ensure position = 0 when propagated

                dsts.add(b);

                leftToFill -= b.remaining();

                accumulatedBytes -= b.remaining();

                if (leftToFill == 0)

                    break;

            } else {

                int prevLimit = b.limit();

                b.limit(b.position() + leftToFill);

                ByteBuffer slice = b.slice();

                dsts.add(slice);

                b.limit(prevLimit);

                b.position(b.position() + leftToFill);

                accumulatedBytes -= leftToFill;

                leftToFill = 0;

                break;

            }

        }

        assert (state == ACTIVE || state == CANCELLED)

                ? leftToFill == 0 : state == COMPLETE;

        assert (state == ACTIVE || state == CANCELLED)

                ? remaining(dsts) == bufferSize : state == COMPLETE;

        assert accumulatedBytes >= 0;

        assert dsts.stream().noneMatch(b -> b.position() != 0);

        return Collections.unmodifiableList(dsts);

    }

    /** Subscription that is passed to the downstream subscriber. */

    private class DownstreamSubscription implements Flow.Subscription {

        private final AtomicBoolean cancelled = new AtomicBoolean(); // false

        private final Demand demand = new Demand();

        private volatile boolean illegalArg;

        @Override

        public void request(long n) {

            if (cancelled.get() || illegalArg) {

                return;

            }

            if (n <= 0L) {

                // pass the "bad" value upstream so the Publisher can deal with

                // it appropriately, i.e. invoke onError

                illegalArg = true;

                subscription.request(n);

                return;

            }

            demand.increase(n);

            pushDemanded();

        }

        private final SequentialScheduler pushDemandedScheduler =

                new SequentialScheduler(new PushDemandedTask());

        void pushDemanded() {

            if (cancelled.get())

                return;

            pushDemandedScheduler.runOrSchedule();

        }

        class PushDemandedTask extends SequentialScheduler.CompleteRestartableTask {

            @Override

            public void run() {

                try {

                    Throwable t = throwable;

                    if (t != null) {

                        pushDemandedScheduler.stop(); // stop the demand scheduler

                        downstreamSubscriber.onError(t);

                        return;

                    }

                    while (true) {

                        List<ByteBuffer> item;

                        synchronized (buffersLock) {

                            if (cancelled.get())

                                return;

                            if (!hasEnoughAccumulatedBytes())

                                break;

                            if (!demand.tryDecrement())

                                break;

                            item = fromInternalBuffers();

                        }

                        assert item != null;

                        downstreamSubscriber.onNext(item);

                    }

                    if (cancelled.get())

                        return;

                    // complete only if all data consumed

                    boolean complete;

                    synchronized (buffersLock) {

                        complete = state == COMPLETE && internalBuffers.isEmpty();

                    }

                    if (complete) {

                        assert internalBuffers.isEmpty();

                        pushDemandedScheduler.stop(); // stop the demand scheduler

                        downstreamSubscriber.onComplete();

                        return;

                    }

                } catch (Throwable t) {

                    cancel();  // cancel if there is any error

                    throw t;

                }

                boolean requestMore = false;

                synchronized (buffersLock) {

                    if (!hasEnoughAccumulatedBytes() && !demand.isFulfilled()) {

                        // request more upstream data

                        requestMore = true;

                    }

                }

                if (requestMore)

                    subscription.request(1);

            }

        }

        @Override

        public void cancel() {

            if (cancelled.compareAndExchange(false, true))

                return;  // already cancelled

            state = CANCELLED;  // set CANCELLED state of upstream subscriber

            subscription.cancel();  // cancel upstream subscription

            pushDemandedScheduler.stop(); // stop the demand scheduler

        }

    }

    @Override

    public void onSubscribe(Flow.Subscription subscription) {

        Objects.requireNonNull(subscription);

        if (this.subscription != null) {

            subscription.cancel();

            return;

        }

        int s = this.state;

        assert s == UNSUBSCRIBED;

        state = ACTIVE;

        this.subscription = subscription;

        downstreamSubscription = new DownstreamSubscription();

        downstreamSubscriber.onSubscribe(downstreamSubscription);

    }

    @Override

    public void onNext(List<ByteBuffer> item) {

        Objects.requireNonNull(item);

        int s = state;

        if (s == CANCELLED)

            return;

        if (s != ACTIVE)

            throw new InternalError("onNext on inactive subscriber");

        synchronized (buffersLock) {

            internalBuffers.addAll(item);

            accumulatedBytes += remaining(item);

        }

        downstreamSubscription.pushDemanded();

    }

    @Override

    public void onError(Throwable incomingThrowable) {

        Objects.requireNonNull(incomingThrowable);

        int s = state;

        assert s == ACTIVE : "Expected ACTIVE, got:" + s;

        state = ERROR;

        Throwable t = this.throwable;

        assert t == null : "Expected null, got:" + t;

        this.throwable = incomingThrowable;

        downstreamSubscription.pushDemanded();

    }

    @Override

    public void onComplete() {

        int s = state;

        assert s == ACTIVE : "Expected ACTIVE, got:" + s;

        state = COMPLETE;

        downstreamSubscription.pushDemanded();

    }

    @Override

    public CompletionStage<T> getBody() {

        return downstreamSubscriber.getBody();

    }

}