Класс 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) упрощает поддержку распространённого сценария, в котором единственное действие подписчика — запрашивать и обрабатывать все элементы с помощью предоставленной функции.

Этот класс также может служить удобным базовым классом для подклассов, которые генерируют элементы и используют методы этого класса для их публикации. Например, ниже приведён класс, который периодически публикует элементы, получаемые от поставщика. (На практике можно добавить методы для независимого запуска и остановки генерации, совместного использования 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) {
    submit(function.apply(item));
    subscription.request(1);
  }
  public void onError(Throwable ex) { closeExceptionally(ex); }
  public void onComplete() { close(); }
}