Класс SubmissionPublisher<T>

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

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

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

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