/*
 * Copyright (c) 2019 Oracle and/or its affiliates. All rights reserved.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License v. 2.0, which is available at
 * http://www.eclipse.org/legal/epl-2.0.
 *
 * This Source Code may also be made available under the following Secondary
 * Licenses when the conditions for such availability set forth in the
 * Eclipse Public License v. 2.0 are satisfied: GNU General Public License,
 * version 2 with the GNU Classpath Exception, which is available at
 * https://www.gnu.org/software/classpath/license.html.
 *
 * SPDX-License-Identifier: EPL-2.0 OR GPL-2.0 WITH Classpath-exception-2.0
 */

package org.glassfish.jersey.internal.jsr166;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.BiPredicate;
import java.util.function.Consumer;

A 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  reactive-streams
Publishers relying on drop handling and/or blocking for flow
control.

A SubmissionPublisher uses the 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 newFixedThreadPool.newFixedThreadPool. Otherwise consider using the default, normally the commonPool.commonPool. 

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 request do not directly result in buffer expansion, but risk saturation if unfilled requests exceed the maximum capacity. The default value of Flow.defaultBufferSize() may provide a useful starting point for choosing a capacity based on expected rates, resources, and usages. 

Publication methods support different policies about what to do when buffers are saturated. Method submit blocks until resources are available. This is simplest, but least responsive. The offer methods may drop items (either immediately or with bounded timeout), but provide an opportunity to interpose a handler and then retry. 

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 onNext, but exceptions in methods onSubscribe, onError and onComplete are not recorded or handled before cancellation. If the supplied Executor throws 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 closeExceptionally in these cases. 

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

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

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

Here is an example of a 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 submit, along with other utility methods. 

 
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(); }
 }
Author:
Doug Lea
Type parameters:
<T> – the published item type
Since:
9/**
 * 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>
 * <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>
 * <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>
 * <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>
 * <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>
 * <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>
 * <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.)
 * <p>
 * <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>
 * <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.
 * <p>
 * <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 SubmittableFlowPublisher<T> {

    private final java.util.concurrent.SubmissionPublisher<T> publisher;

    
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 onNext. 
Params:
executor –          the executor to use for async delivery,
                         supporting creation of at least one independent thread
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 getMaxBufferCapacity returns the actual value)
handler –  if non-null, procedure to invoke upon exception thrown in method onNext
Throws:
NullPointerException –     if executor is null
IllegalArgumentException – if maxBufferCapacity not
                                 positive/**
     * 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) {
        publisher = new java.util.concurrent.SubmissionPublisher<T>(executor, maxBufferCapacity, convertConsumer(handler));
    }

    
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 onNext. 
Params:
executor –          the executor to use for async delivery,
                         supporting creation of at least one independent thread
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 getMaxBufferCapacity returns the actual value)
Throws:
NullPointerException –     if executor is null
IllegalArgumentException – if maxBufferCapacity not
                                 positive/**
     * 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) {
        publisher = new java.util.concurrent.SubmissionPublisher<T>(executor, maxBufferCapacity);
    }

    
Creates a new SubmissionPublisher using the commonPool.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 Flow.defaultBufferSize, and no handler for Subscriber exceptions in method onNext. /**
     * 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() {
        publisher = new java.util.concurrent.SubmissionPublisher<T>();
    }

    
{@inheritDoc}
/**
     * {@inheritDoc}
     */
    @Override
    public CompletableFuture<Void> consume(Consumer<? super T> consumer) {
        return publisher.consume(consumer);
    }

    
{@inheritDoc}
/**
     * {@inheritDoc}
     */
    @Override
    public void close() {
        publisher.close();
    }

    
{@inheritDoc}
/**
     * {@inheritDoc}
     */
    @Override
    public void closeExceptionally(Throwable error) {
        publisher.closeExceptionally(error);
    }

    
{@inheritDoc}
/**
     * {@inheritDoc}
     */
    @Override
    public long estimateMinimumDemand() {
        return publisher.estimateMinimumDemand();
    }

    
{@inheritDoc}
/**
     * {@inheritDoc}
     */
    @Override
    public int estimateMaximumLag() {
        return publisher.estimateMaximumLag();
    }

    
{@inheritDoc}
/**
     * {@inheritDoc}
     */
    @Override
    public Throwable getClosedException() {
        return publisher.getClosedException();
    }

    
{@inheritDoc}
/**
     * {@inheritDoc}
     */
    @Override
    public int getMaxBufferCapacity() {
        return publisher.getMaxBufferCapacity();
    }

    
{@inheritDoc}
/**
     * {@inheritDoc}
     */
    @Override
    public int offer(T item, long timeout, TimeUnit unit, BiPredicate<Flow.Subscriber<? super T>, ? super T> onDrop) {
        return publisher.offer(item, timeout, unit, convertPredicate(onDrop));
    }

    
{@inheritDoc}
/**
     * {@inheritDoc}
     */
    public int offer(T item, BiPredicate<Flow.Subscriber<? super T>, ? super T> onDrop) {
        return publisher.offer(item, convertPredicate(onDrop));
    }

    
{@inheritDoc}
/**
     * {@inheritDoc}
     */
    @Override
    public int submit(T item) {
        return publisher.submit(item);
    }

    
{@inheritDoc}
/**
     * {@inheritDoc}
     */
    @Override
    public void subscribe(Flow.Subscriber<? super T> subscriber) {
        publisher.subscribe(convertSubscriber(subscriber));
    }

    private static <T> BiConsumer<? super java.util.concurrent.Flow.Subscriber<? super T>, ? super Throwable> convertConsumer(
            BiConsumer<? super Flow.Subscriber<? super T>, ? super Throwable> consumer) {
        return new BiConsumer<java.util.concurrent.Flow.Subscriber<? super T>, Throwable>() {
            @Override
            public void accept(java.util.concurrent.Flow.Subscriber<? super T> subscriber, Throwable throwable) {
                consumer.accept(convertSubscriber(subscriber), throwable);
            }
        };
    }

    private static <T> BiPredicate<java.util.concurrent.Flow.Subscriber<? super T>, ? super T> convertPredicate(
            BiPredicate<Flow.Subscriber<? super T>, ? super T> predicate) {
        return new BiPredicate<java.util.concurrent.Flow.Subscriber<? super T>, T>() {
            @Override
            public boolean test(java.util.concurrent.Flow.Subscriber<? super T> subscriber, T t) {
                return predicate.test(convertSubscriber(subscriber), t);
            }
        };
    }

    private static <T> Flow.Subscriber<? super T> convertSubscriber(java.util.concurrent.Flow.Subscriber<? super T> subscriber) {
        return new Flow.Subscriber<T>() {

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                subscriber.onSubscribe(convertSubscription(subscription));
            }

            @Override
            public void onNext(T item) {
                subscriber.onNext(item);
            }

            @Override
            public void onError(Throwable throwable) {
                subscriber.onError(throwable);
            }

            @Override
            public void onComplete() {
                subscriber.onComplete();
            }
        };
    }

    private static <T> java.util.concurrent.Flow.Subscriber<T> convertSubscriber(Flow.Subscriber<T> subscriber) {
        return new java.util.concurrent.Flow.Subscriber<T>() {

            @Override
            public void onSubscribe(java.util.concurrent.Flow.Subscription subscription) {
                subscriber.onSubscribe(convertSubscription(subscription));
            }

            @Override
            public void onNext(T item) {
                subscriber.onNext(item);
            }

            @Override
            public void onError(Throwable throwable) {
                subscriber.onError(throwable);
            }

            @Override
            public void onComplete() {
                subscriber.onComplete();
            }
        };
    }

    private static java.util.concurrent.Flow.Subscription convertSubscription(Flow.Subscription subscription) {
        return new java.util.concurrent.Flow.Subscription() {
            @Override
            public void request(long n) {
                subscription.request(n);
            }

            @Override
            public void cancel() {
                subscription.cancel();
            }
        };
    }

    private static Flow.Subscription convertSubscription(java.util.concurrent.Flow.Subscription subscription) {
        return new Flow.Subscription() {
            @Override
            public void request(long n) {
                subscription.request(n);
            }

            @Override
            public void cancel() {
                subscription.cancel();
            }
        };
    }
}