Класс SubmissionPublisher<T>

public class SubmissionPublisher<T>
extends Object
implements Flow.Publisher<T>, AutoCloseable

A Flow.Publisher который асинхронно выдает отправленные (не пустые) элементы текущим подписчикам до его закрытия. Каждый текущий подписчик получает новые отправленные элементы в том же порядке, если не возникают ошибки или исключения. Использование SubmissionPublisher позволяет генераторам элементов действовать как совместимые reactive-streams Publishers, полагаясь на обработку ошибок и/или блокировку для управления потоком.

SubmissionPublisher использует Executor, предоставленный в его конструкторе для доставки подписчикам. Лучший выбор Executor зависит от ожидаемого использования. Если генератор(ы) отправляемых элементов работают в отдельных потоках, и количество подписчиков можно оценить, рассмотрите использование Executors.newFixedThreadPool(int). В противном случае рассмотрите использование по умолчанию, обычно ForkJoinPool.commonPool().

Буферизация позволяет производителям и потребителям временно работать с разными скоростями. Каждый подписчик использует независимый буфер. Буферы создаются при первом использовании и расширяются по мере необходимости до заданного максимума. (Принудительная емкость может быть округлена до ближайшей степени двойки и/или ограничена наибольшим значением, поддерживаемым этой реализацией.) Вызовы request не напрямую приводят к расширению буфера, но рискуют насыщением, если необработанные запросы превысят максимальную емкость. Значение по умолчанию Flow.defaultBufferSize() может служить хорошей отправной точкой для выбора емкости на основе ожидаемых скоростей, ресурсов и использования.

Один SubmissionPublisher может быть разделен между несколькими источниками. Действия в потоке источника перед публикацией элемента или выдачей сигнала happen-before последующие действия соответствующего доступа каждого подписчика. Однако указанные оценки задержки и спроса предназначены для мониторинга, а не для управления синхронизацией и могут отражать устаревшие или неточные представления о ходе работы.

Методы публикации поддерживают различные политики относительно того, что делать, когда буферы заполнены. Метод submit блокируется до тех пор, пока ресурсы не станут доступными. Это проще, но менее отзывчиво. Методы offer могут отбрасывать элементы (немедленно или с ограниченным таймаутом), но предоставляют возможность вставить обработчик, а затем повторить попытку.

Если какой-либо метод Subscriber вызывает исключение, его подписка отменяется. Если обработчик передан в качестве аргумента конструктора, он вызывается перед отменой при возникновении исключения в методе onNext, но исключения в методах onSubscribe, onError и onComplete не регистрируются и не обрабатываются перед отменой. Если предоставленный Executor вызывает RejectedExecutionException (или любое другое RuntimeException или Error) при попытке выполнить задачу, или обработчик отбрасывания вызывает исключение при обработке отброшенного элемента, то исключение повторно выбрасывается. В этих случаях не все подписчики получат опубликованный элемент. В таких случаях обычно рекомендуется closeExceptionally.

Метод consume(Consumer) упрощает поддержку распространённого случая, в котором единственным действием подписчика является запрос и обработка всех элементов с использованием предоставленной функции.

Этот класс также может служить удобной основой для подклассов, которые генерируют элементы и используют методы этого класса для их публикации. Например, вот класс, который периодически публикует элементы, сгенерированные из поставщика. (На практике вы можете добавить методы для независимого запуска и остановки генерации, для совместного использования Executors между издателями и так далее, или использовать SubmissionPublisher в качестве компонента, а не суперкласса.)

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

Вот пример реализации Flow.Processor. Он использует запросы с одним шагом к своему издателю для простоты иллюстрации. Более адаптивная версия могла бы отслеживать поток, используя оценку задержки, возвращаемую от submit, вместе с другими служебными методами.

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