Как связать нереактивный Spring EventListener и реактивный Flux

В чем разница между созданием потока напрямую путем вызова Flux.push и использованием приемника в выражении лямбады push по сравнению с использованием приемника, предоставленного DirectProcessor?

В минимальном примере, где Flux просто генерирует пару событий, я мог бы сделать

Flux.<String>push(emitter -> {
   emitter.next("One");
   emitter.next("Two");
   emitter.complete();
 });

по сравнению с использованием DirectProcessor

var emitter = DirectProcessor.<String>create().sink();
emitter.next("One");
emitter.next("Two");
emitter.complete();

Просто чтобы уточнить: я знаю, что могу использовать здесь Flux.just, но мой вариант использования на самом деле строит мост между Spring @EventListeners и Spring WebFlux, где я хочу создать Flux для каждого входящего запроса SSE для определенного ресурса, а затем публиковать события в этот Flux.

Может ли кто-нибудь сказать мне, действительны ли оба подхода? Конечно, разница должна быть. В частности, раздел Справочное руководство по Reactor на DirectProcessor гласит:

С другой стороны, у него есть ограничение на отсутствие противодавления. Как следствие, DirectProcessor сигнализирует своим подписчикам об исключении IllegalStateException, если вы проталкиваете через него N элементов, но по крайней мере один из его подписчиков запросил меньше N.

Что это обозначает?

[EDIT:] В более ранней версии вопроса я использовал Flux.generate() вместо Flux.push(), что, очевидно, неверно, поскольку сгенерировать может создать не более одного события.

[EDIT 2:] @ 123 попросил меня привести полный пример того, чего я пытаюсь достичь. Потерпите меня, это изрядное количество кода для SO-вопроса:

Полный пример того, что я на самом деле пытаюсь сделать

Я хотел бы построить мост между (нереактивным) прослушивателем событий домена Spring и реактивным Flux, который я затем могу использовать в конечной точке WebFlux для публикации SSE. В следующих фрагментах кода для краткости используются аннотации Lombok.

Предположим, что я в конечном итоге хочу опубликовать состояние пользователя в процессе адаптации как SSE. Вот перечисление:

public enum ProcessState {
  CREATED(false),
  VERIFIED(false),
  AUTHORIZATION_PENDING(false),
  AUTHORIZED(false),
  ACTIVE(true);

  @Getter
  private final boolean terminalState;

  ProcessState(boolean terminalState) {
    this.terminalState = terminalState;
  }

}

Нереактивная бизнес-логика будет публиковать StateChangedEvents всякий раз, когда изменяется состояние любого пользователя:

@Data
@RequiredArgsConstructor
public class StateChangedEvent {
  private final UUID userId;
  private final ProcessState newState;
}

Отсюда и мой первоначальный вопрос. Как мне построить мост, который переводит события этой области в поток Flux? Мои требования:

• Текущее состояние процесса должно быть передано, как только новый клиент зарегистрируется.
• Поток Flux должен завершаться всякий раз, когда достигается "терминальное" состояние подключения.

Вот что у меня есть на данный момент:

@Component
@RequiredArgsConstructor
class EventBridge {

  @RequiredArgsConstructor(access = PRIVATE)
  private static class Subscriber {
    private final UUID userId;
    private final FluxSink<ProcessState> sink;
    private boolean eventEmitted;
  }

  private final UserRepository repository;
  private final Map<UUID, Subscriber> subscribers = new ConcurrentHashMap<>();

  @EventListener
  void stateChanged(StateChangedEvent event) {
    notifySubscribers(event);
  }

  Flux<ProcessState> register(UUID userId) {
    return Flux.push(emitter -> addSubscriber(userId, emitter));
  }

  private Subscriber addSubscriber(UUID userId, FluxSink<ProcessState> sink) {
    var subscriptionId = randomUUID();
    var subscriber = new Subscriber(userId, sink);
    subscribers.put(subscriptionId, subscriber);
    sink
      .onRequest(n -> poll(subscriber))
      .onDispose(() -> removeSubscriber(subscriptionId));
    return subscriber;
  }

  private void poll(Subscriber subscriber) {
    emit(subscriber, loadCurrentState(subscriber), true);
  }

  private ProcessState loadCurrentState(Subscriber subscriber) {
    return repository.findById(subscriber.userId).getProcessState();
  }

  private void removeSubscriber(UUID subscriptionId) {
    subscribers.remove(subscriptionId);
  }

  private void notifySubscribers(StateChangedEvent event) {
    subscribers.values().stream()
      .filter(subscriber -> subscriber.userId.equals(event.getUserId()))
      .forEach(subscriber -> emit(subscriber, event.getNewState(), false));
  }

  private void emit(Subscriber subscriber, ProcessState processState, boolean onlyIfFirst) {
    synchronized (subscriber) {
      if (onlyIfFirst && subscriber.eventEmitted) {
        return;
      }
      subscriber.sink.next(processState);
      if (processState.isTerminalState()) {
        subscriber.sink.complete();
      }
      subscriber.eventEmitted = true;
    }
  }

}

И, наконец, контроллер, на котором используется мост:

@RestController
@RequiredArgsConstructor
class UserController {

  private final EventBridge eventBridge;

  @GetMapping(value = "/{userId}", produces = TEXT_EVENT_STREAM_VALUE)
  Flux<ServerSentEvent<ProcessState>> readAsStream(@PathVariable UUID userId) {
    return eventBridge.register(userId).map(response -> ServerSentEvent.builder((ProcessState) response).build());
  }

}

В моем коде моста есть пара проблем, которые я не могу понять:

• Мне действительно нужно синхронизировать свой Subscriber экземпляр, чтобы избежать записи устаревших событий из poll исходного состояния? Если я этого не сделаю, произойдет так, что событие StateChange прибывает и публикуется до чтения текущего состояния из репозитория, который затем выталкивается не по порядку. Конечно, должен существовать более элегантный способ решения этой проблемы в стиле Flux без ключевого слова synchronized.

• Мы уже исключили Flux.generate, похоже, это работает с Flux.push, Flux.create будет генерировать намного больше событий SSE? Почему? Боюсь, я не понимаю различий между этими тремя.

• Вместо того, чтобы использовать статические методы на Flux, следует ли мне использовать здесь DirectProcessor или любой другой процессор? Я новичок во всем реактивном стеке, и документация Spring Reactor для меня слишком расплывчата, tbh. Еще раз: в чем разница? Как насчет того комментария о противодавлении, о котором я упоминал выше?