/*

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

 */

/*

 * This file is available under and governed by the GNU General Public

 * License version 2 only, as published by the Free Software Foundation.

 * However, the following notice accompanied the original version of this

 * file:

 *

 * Written by Doug Lea with assistance from members of JCP JSR-166

 * Expert Group and released to the public domain, as explained at

 * http://creativecommons.org/publicdomain/zero/1.0/

 */

package java.util.concurrent;

import java.util.ArrayList;

import java.util.List;

import java.util.concurrent.locks.LockSupport;

import java.util.function.BiConsumer;

import java.util.function.BiPredicate;

import java.util.function.Consumer;

/**

 * A {@link Flow.Publisher} that asynchronously issues submitted

 * (non-null) items to current subscribers until it is closed.  Each

 * current subscriber receives newly submitted items in the same order

 * unless drops or exceptions are encountered.  Using a

 * SubmissionPublisher allows item generators to act as compliant <a

 * href="http://www.reactive-streams.org/"> reactive-streams</a>

 * Publishers relying on drop handling and/or blocking for flow

 * control.

 *

 * <p>A SubmissionPublisher uses the {@link Executor} supplied in its

 * constructor for delivery to subscribers. The best choice of

 * Executor depends on expected usage. If the generator(s) of

 * submitted items run in separate threads, and the number of

 * subscribers can be estimated, consider using a {@link

 * Executors#newFixedThreadPool}. Otherwise consider using the

 * default, normally the {@link ForkJoinPool#commonPool}.

 *

 * <p>Buffering allows producers and consumers to transiently operate

 * at different rates.  Each subscriber uses an independent buffer.

 * Buffers are created upon first use and expanded as needed up to the

 * given maximum. (The enforced capacity may be rounded up to the

 * nearest power of two and/or bounded by the largest value supported

 * by this implementation.)  Invocations of {@link

 * Flow.Subscription#request(long) request} do not directly result in

 * buffer expansion, but risk saturation if unfilled requests exceed

 * the maximum capacity.  The default value of {@link

 * Flow#defaultBufferSize()} may provide a useful starting point for

 * choosing a capacity based on expected rates, resources, and usages.

 *

 * <p>Publication methods support different policies about what to do

 * when buffers are saturated. Method {@link #submit(Object) submit}

 * blocks until resources are available. This is simplest, but least

 * responsive.  The {@code offer} methods may drop items (either

 * immediately or with bounded timeout), but provide an opportunity to

 * interpose a handler and then retry.

 *

 * <p>If any Subscriber method throws an exception, its subscription

 * is cancelled.  If a handler is supplied as a constructor argument,

 * it is invoked before cancellation upon an exception in method

 * {@link Flow.Subscriber#onNext onNext}, but exceptions in methods

 * {@link Flow.Subscriber#onSubscribe onSubscribe},

 * {@link Flow.Subscriber#onError(Throwable) onError} and

 * {@link Flow.Subscriber#onComplete() onComplete} are not recorded or

 * handled before cancellation.  If the supplied Executor throws

 * {@link RejectedExecutionException} (or any other RuntimeException

 * or Error) when attempting to execute a task, or a drop handler

 * throws an exception when processing a dropped item, then the

 * exception is rethrown. In these cases, not all subscribers will

 * have been issued the published item. It is usually good practice to

 * {@link #closeExceptionally closeExceptionally} in these cases.

 *

 * <p>Method {@link #consume(Consumer)} simplifies support for a

 * common case in which the only action of a subscriber is to request

 * and process all items using a supplied function.

 *

 * <p>This class may also serve as a convenient base for subclasses

 * that generate items, and use the methods in this class to publish

 * them.  For example here is a class that periodically publishes the

 * items generated from a supplier. (In practice you might add methods

 * to independently start and stop generation, to share Executors

 * among publishers, and so on, or use a SubmissionPublisher as a

 * component rather than a superclass.)

 *

 * <pre> {@code

 * class PeriodicPublisher<T> extends SubmissionPublisher<T> {

 *   final ScheduledFuture<?> periodicTask;

 *   final ScheduledExecutorService scheduler;

 *   PeriodicPublisher(Executor executor, int maxBufferCapacity,

 *                     Supplier<? extends T> supplier,

 *                     long period, TimeUnit unit) {

 *     super(executor, maxBufferCapacity);

 *     scheduler = new ScheduledThreadPoolExecutor(1);

 *     periodicTask = scheduler.scheduleAtFixedRate(

 *       () -> submit(supplier.get()), 0, period, unit);

 *   }

 *   public void close() {

 *     periodicTask.cancel(false);

 *     scheduler.shutdown();

 *     super.close();

 *   }

 * }}</pre>

 *

 * <p>Here is an example of a {@link Flow.Processor} implementation.

 * It uses single-step requests to its publisher for simplicity of

 * illustration. A more adaptive version could monitor flow using the

 * lag estimate returned from {@code submit}, along with other utility

 * methods.

 *

 * <pre> {@code

 * class TransformProcessor<S,T> extends SubmissionPublisher<T>

 *   implements Flow.Processor<S,T> {

 *   final Function<? super S, ? extends T> function;

 *   Flow.Subscription subscription;

 *   TransformProcessor(Executor executor, int maxBufferCapacity,

 *                      Function<? super S, ? extends T> function) {

 *     super(executor, maxBufferCapacity);

 *     this.function = function;

 *   }

 *   public void onSubscribe(Flow.Subscription subscription) {

 *     (this.subscription = subscription).request(1);

 *   }

 *   public void onNext(S item) {

 *     subscription.request(1);

 *     submit(function.apply(item));

 *   }

 *   public void onError(Throwable ex) { closeExceptionally(ex); }

 *   public void onComplete() { close(); }

 * }}</pre>

 *

 * @param <T> the published item type

 * @author Doug Lea

 * @since 9

 */

public class SubmissionPublisher<T> implements Flow.Publisher<T>,

                                               AutoCloseable {

    /*

     * Most mechanics are handled by BufferedSubscription. This class

     * mainly tracks subscribers and ensures sequentiality, by using

     * built-in synchronization locks across public methods. (Using

     * built-in locks works well in the most typical case in which

     * only one thread submits items).

     */

    /** The largest possible power of two array size. */

    static final int BUFFER_CAPACITY_LIMIT = 1 << 30;

    /** Round capacity to power of 2, at most limit. */

    static final int roundCapacity(int cap) {

        int n = cap - 1;

        n |= n >>> 1;

        n |= n >>> 2;

        n |= n >>> 4;

        n |= n >>> 8;

        n |= n >>> 16;

        return (n <= 0) ? 1 : // at least 1

            (n >= BUFFER_CAPACITY_LIMIT) ? BUFFER_CAPACITY_LIMIT : n + 1;

    }

    // default Executor setup; nearly the same as CompletableFuture

    /**

     * Default executor -- ForkJoinPool.commonPool() unless it cannot

     * support parallelism.

     */

    private static final Executor ASYNC_POOL =

        (ForkJoinPool.getCommonPoolParallelism() > 1) ?

        ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();

    /** Fallback if ForkJoinPool.commonPool() cannot support parallelism */

    private static final class ThreadPerTaskExecutor implements Executor {

        public void execute(Runnable r) { new Thread(r).start(); }

    }

    /**

     * Clients (BufferedSubscriptions) are maintained in a linked list

     * (via their "next" fields). This works well for publish loops.

     * It requires O(n) traversal to check for duplicate subscribers,

     * but we expect that subscribing is much less common than

     * publishing. Unsubscribing occurs only during traversal loops,

     * when BufferedSubscription methods return negative values

     * signifying that they have been disabled.  To reduce

     * head-of-line blocking, submit and offer methods first call

     * BufferedSubscription.offer on each subscriber, and place

     * saturated ones in retries list (using nextRetry field), and

     * retry, possibly blocking or dropping.

     */

    BufferedSubscription<T> clients;

    /** Run status, updated only within locks */

    volatile boolean closed;

    /** If non-null, the exception in closeExceptionally */

    volatile Throwable closedException;

    // Parameters for constructing BufferedSubscriptions

    final Executor executor;

    final BiConsumer<? super Flow.Subscriber<? super T>, ? super Throwable> onNextHandler;

    final int maxBufferCapacity;

    /**

     * Creates a new SubmissionPublisher using the given Executor for

     * async delivery to subscribers, with the given maximum buffer size

     * for each subscriber, and, if non-null, the given handler invoked

     * when any Subscriber throws an exception in method {@link

     * Flow.Subscriber#onNext(Object) onNext}.

     *

     * @param executor the executor to use for async delivery,

     * supporting creation of at least one independent thread

     * @param maxBufferCapacity the maximum capacity for each

     * subscriber's buffer (the enforced capacity may be rounded up to

     * the nearest power of two and/or bounded by the largest value

     * supported by this implementation; method {@link #getMaxBufferCapacity}

     * returns the actual value)

     * @param handler if non-null, procedure to invoke upon exception

     * thrown in method {@code onNext}

     * @throws NullPointerException if executor is null

     * @throws IllegalArgumentException if maxBufferCapacity not

     * positive

     */

    public SubmissionPublisher(Executor executor, int maxBufferCapacity,

                               BiConsumer<? super Flow.Subscriber<? super T>, ? super Throwable> handler) {

        if (executor == null)

            throw new NullPointerException();

        if (maxBufferCapacity <= 0)

            throw new IllegalArgumentException("capacity must be positive");

        this.executor = executor;

        this.onNextHandler = handler;

        this.maxBufferCapacity = roundCapacity(maxBufferCapacity);

    }

    /**

     * Creates a new SubmissionPublisher using the given Executor for

     * async delivery to subscribers, with the given maximum buffer size

     * for each subscriber, and no handler for Subscriber exceptions in

     * method {@link Flow.Subscriber#onNext(Object) onNext}.

     *

     * @param executor the executor to use for async delivery,

     * supporting creation of at least one independent thread

     * @param maxBufferCapacity the maximum capacity for each

     * subscriber's buffer (the enforced capacity may be rounded up to

     * the nearest power of two and/or bounded by the largest value

     * supported by this implementation; method {@link #getMaxBufferCapacity}

     * returns the actual value)

     * @throws NullPointerException if executor is null

     * @throws IllegalArgumentException if maxBufferCapacity not

     * positive

     */

    public SubmissionPublisher(Executor executor, int maxBufferCapacity) {

        this(executor, maxBufferCapacity, null);

    }

    /**

     * Creates a new SubmissionPublisher using the {@link

     * ForkJoinPool#commonPool()} for async delivery to subscribers

     * (unless it does not support a parallelism level of at least two,

     * in which case, a new Thread is created to run each task), with

     * maximum buffer capacity of {@link Flow#defaultBufferSize}, and no

     * handler for Subscriber exceptions in method {@link

     * Flow.Subscriber#onNext(Object) onNext}.

     */

    public SubmissionPublisher() {

        this(ASYNC_POOL, Flow.defaultBufferSize(), null);

    }

    /**

     * Adds the given Subscriber unless already subscribed.  If already

     * subscribed, the Subscriber's {@link

     * Flow.Subscriber#onError(Throwable) onError} method is invoked on

     * the existing subscription with an {@link IllegalStateException}.

     * Otherwise, upon success, the Subscriber's {@link

     * Flow.Subscriber#onSubscribe onSubscribe} method is invoked

     * asynchronously with a new {@link Flow.Subscription}.  If {@link

     * Flow.Subscriber#onSubscribe onSubscribe} throws an exception, the

     * subscription is cancelled. Otherwise, if this SubmissionPublisher

     * was closed exceptionally, then the subscriber's {@link

     * Flow.Subscriber#onError onError} method is invoked with the

     * corresponding exception, or if closed without exception, the

     * subscriber's {@link Flow.Subscriber#onComplete() onComplete}

     * method is invoked.  Subscribers may enable receiving items by

     * invoking the {@link Flow.Subscription#request(long) request}

     * method of the new Subscription, and may unsubscribe by invoking

     * its {@link Flow.Subscription#cancel() cancel} method.

     *

     * @param subscriber the subscriber

     * @throws NullPointerException if subscriber is null

     */

    public void subscribe(Flow.Subscriber<? super T> subscriber) {

        if (subscriber == null) throw new NullPointerException();

        BufferedSubscription<T> subscription =

            new BufferedSubscription<T>(subscriber, executor,

                                        onNextHandler, maxBufferCapacity);

        synchronized (this) {

            for (BufferedSubscription<T> b = clients, pred = null;;) {

                if (b == null) {

                    Throwable ex;

                    subscription.onSubscribe();

                    if ((ex = closedException) != null)

                        subscription.onError(ex);

                    else if (closed)

                        subscription.onComplete();

                    else if (pred == null)

                        clients = subscription;

                    else

                        pred.next = subscription;

                    break;

                }

                BufferedSubscription<T> next = b.next;

                if (b.isDisabled()) { // remove

                    b.next = null;    // detach

                    if (pred == null)

                        clients = next;

                    else

                        pred.next = next;

                }

                else if (subscriber.equals(b.subscriber)) {

                    b.onError(new IllegalStateException("Duplicate subscribe"));

                    break;

                }

                else

                    pred = b;

                b = next;

            }

        }

    }

    /**

     * Publishes the given item to each current subscriber by

     * asynchronously invoking its {@link Flow.Subscriber#onNext(Object)

     * onNext} method, blocking uninterruptibly while resources for any

     * subscriber are unavailable. This method returns an estimate of

     * the maximum lag (number of items submitted but not yet consumed)

     * among all current subscribers. This value is at least one

     * (accounting for this submitted item) if there are any

     * subscribers, else zero.

     *

     * <p>If the Executor for this publisher throws a

     * RejectedExecutionException (or any other RuntimeException or

     * Error) when attempting to asynchronously notify subscribers,

     * then this exception is rethrown, in which case not all

     * subscribers will have been issued this item.

     *

     * @param item the (non-null) item to publish

     * @return the estimated maximum lag among subscribers

     * @throws IllegalStateException if closed

     * @throws NullPointerException if item is null

     * @throws RejectedExecutionException if thrown by Executor

     */

    public int submit(T item) {

        if (item == null) throw new NullPointerException();

        int lag = 0;

        boolean complete;

        synchronized (this) {

            complete = closed;

            BufferedSubscription<T> b = clients;

            if (!complete) {

                BufferedSubscription<T> pred = null, r = null, rtail = null;

                while (b != null) {

                    BufferedSubscription<T> next = b.next;

                    int stat = b.offer(item);

                    if (stat < 0) {           // disabled

                        b.next = null;

                        if (pred == null)

                            clients = next;

                        else

                            pred.next = next;

                    }

                    else {

                        if (stat > lag)

                            lag = stat;

                        else if (stat == 0) { // place on retry list

                            b.nextRetry = null;

                            if (rtail == null)

                                r = b;

                            else

                                rtail.nextRetry = b;

                            rtail = b;

                        }

                        pred = b;

                    }

                    b = next;

                }

                while (r != null) {

                    BufferedSubscription<T> nextRetry = r.nextRetry;

                    r.nextRetry = null;

                    int stat = r.submit(item);

                    if (stat > lag)

                        lag = stat;

                    else if (stat < 0 && clients == r)

                        clients = r.next; // postpone internal unsubscribes

                    r = nextRetry;

                }

            }

        }

        if (complete)

            throw new IllegalStateException("Closed");

        else

            return lag;

    }

    /**

     * Publishes the given item, if possible, to each current subscriber

     * by asynchronously invoking its {@link

     * Flow.Subscriber#onNext(Object) onNext} method. The item may be

     * dropped by one or more subscribers if resource limits are

     * exceeded, in which case the given handler (if non-null) is

     * invoked, and if it returns true, retried once.  Other calls to

     * methods in this class by other threads are blocked while the

     * handler is invoked.  Unless recovery is assured, options are

     * usually limited to logging the error and/or issuing an {@link

     * Flow.Subscriber#onError(Throwable) onError} signal to the

     * subscriber.

     *

     * <p>This method returns a status indicator: If negative, it

     * represents the (negative) number of drops (failed attempts to

     * issue the item to a subscriber). Otherwise it is an estimate of

     * the maximum lag (number of items submitted but not yet

     * consumed) among all current subscribers. This value is at least

     * one (accounting for this submitted item) if there are any

     * subscribers, else zero.

     *

     * <p>If the Executor for this publisher throws a

     * RejectedExecutionException (or any other RuntimeException or

     * Error) when attempting to asynchronously notify subscribers, or

     * the drop handler throws an exception when processing a dropped

     * item, then this exception is rethrown.

     *

     * @param item the (non-null) item to publish

     * @param onDrop if non-null, the handler invoked upon a drop to a

     * subscriber, with arguments of the subscriber and item; if it

     * returns true, an offer is re-attempted (once)

     * @return if negative, the (negative) number of drops; otherwise

     * an estimate of maximum lag

     * @throws IllegalStateException if closed

     * @throws NullPointerException if item is null

     * @throws RejectedExecutionException if thrown by Executor

     */

    public int offer(T item,

                     BiPredicate<Flow.Subscriber<? super T>, ? super T> onDrop) {

        return doOffer(0L, item, onDrop);

    }

    /**

     * Publishes the given item, if possible, to each current subscriber

     * by asynchronously invoking its {@link

     * Flow.Subscriber#onNext(Object) onNext} method, blocking while

     * resources for any subscription are unavailable, up to the

     * specified timeout or until the caller thread is interrupted, at

     * which point the given handler (if non-null) is invoked, and if it

     * returns true, retried once. (The drop handler may distinguish

     * timeouts from interrupts by checking whether the current thread

     * is interrupted.)  Other calls to methods in this class by other

     * threads are blocked while the handler is invoked.  Unless

     * recovery is assured, options are usually limited to logging the

     * error and/or issuing an {@link Flow.Subscriber#onError(Throwable)

     * onError} signal to the subscriber.

     *

     * <p>This method returns a status indicator: If negative, it

     * represents the (negative) number of drops (failed attempts to

     * issue the item to a subscriber). Otherwise it is an estimate of

     * the maximum lag (number of items submitted but not yet

     * consumed) among all current subscribers. This value is at least

     * one (accounting for this submitted item) if there are any

     * subscribers, else zero.

     *

     * <p>If the Executor for this publisher throws a

     * RejectedExecutionException (or any other RuntimeException or

     * Error) when attempting to asynchronously notify subscribers, or

     * the drop handler throws an exception when processing a dropped

     * item, then this exception is rethrown.

     *

     * @param item the (non-null) item to publish

     * @param timeout how long to wait for resources for any subscriber

     * before giving up, in units of {@code unit}

     * @param unit a {@code TimeUnit} determining how to interpret the

     * {@code timeout} parameter

     * @param onDrop if non-null, the handler invoked upon a drop to a

     * subscriber, with arguments of the subscriber and item; if it

     * returns true, an offer is re-attempted (once)

     * @return if negative, the (negative) number of drops; otherwise

     * an estimate of maximum lag

     * @throws IllegalStateException if closed

     * @throws NullPointerException if item is null

     * @throws RejectedExecutionException if thrown by Executor

     */

    public int offer(T item, long timeout, TimeUnit unit,

                     BiPredicate<Flow.Subscriber<? super T>, ? super T> onDrop) {

        return doOffer(unit.toNanos(timeout), item, onDrop);

    }

    /** Common implementation for both forms of offer */

    final int doOffer(long nanos, T item,

                      BiPredicate<Flow.Subscriber<? super T>, ? super T> onDrop) {

        if (item == null) throw new NullPointerException();

        int lag = 0, drops = 0;

        boolean complete;

        synchronized (this) {

            complete = closed;

            BufferedSubscription<T> b = clients;

            if (!complete) {

                BufferedSubscription<T> pred = null, r = null, rtail = null;

                while (b != null) {

                    BufferedSubscription<T> next = b.next;