Класс SubmissionPublisher<T>

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

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

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

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

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

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

Этот класс также может служить удобной основой для подклассов, генерирующих элементы и использующих методы этого класса для их публикации. Например, вот класс, который периодически публикует элементы, сгенерированные из поставщика. (На практике вы можете добавить методы для независимого запуска и остановки генерации, для совместного использования Executor между издателями и так далее, или использовать 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(); }
 }