Класс SubmissionPublisher<T>

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

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

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

Методы публикации поддерживают различные стратегии действий при переполнении буферов. Метод 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(); }
 }