Как установить мост между нереактивным 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(), что, очевидно, неверно, потому что generate может создать не более одного события.
[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. Опять же: в чем разница? Как насчет того комментария о противодавлении, о котором я упоминал выше?
2 ответа
Итак, если я понимаю, что вы пытаетесь сделать правильно, я думаю, что ваше решение можно было бы значительно упростить.
@Component
public class EventBridge {
private final UserRepository repository;
private final ReplayProcessor<StateChangedEvent> processor;
private final FluxSink<StateChangedEvent> sink;
EventBridge(UserRepository repository){
this.repository= repository;
//Replays events from last 100S for every new subscriber
this.processor = ReplayProcessor.createTimeout(Duration.ofSeconds(100L));
//Sink provides thread safe next,complete and error for subscribers
this.sink = processor.sink();
}
public void changeState(StateChangedEvent event) {
//Literally just pass event into sink, calls onNext on subscribers
sink.next(event);
}
public Flux<ProcessState> streamProcessStateForUser(UUID uuid){
return
//Search repository first, this isn't great since it blocks until
//the repo returns, although that seems to be what you want
//Also I added an Unknown to ProcessState, since it's better than
//it being null.
//Also you should probably return optional from repo.
Flux.concat(
Flux.just(
userRepo.findById(uuid).map(User::getProcessState).orElse(ProcessState.UNKNOWN)
),
processor
//Check the uuid matches the event
.filter(stateChangedEvent -> stateChangedEvent.getUserId().equals(uuid))
//Time out after 100 seconds, not needed but may be useful for you
.take(Duration.ofSeconds(100L))
//Complete flux when at terminal state
.takeUntil(stateChangedEvent -> stateChangedEvent.getNewState().isTerminalState())
//Convert to ProcessState from StateChangedEvent
.map(StateChangedEvent::getNewState)
);
}
}
Должен быть в состоянии сохранить все остальное без изменений.
Спасибо @123 за ответ на мой вопрос о том, как построить мост между Spring @EventListener и Flux. Как упоминалось в вопросе, полный вариант использования заключался в отправке событий домена клиенту с использованием поддержки SSE WebFlux.
Немного подумав, я понял, что вообще не имеет смысла строить этот мост, потому что в сценарии с несколькими экземплярами HTTP-запрос может попасть в другой экземпляр, чем тот, где выполняется процесс адаптации, и, следовательно, никакие события не будут отправляться.
Итак, в конце концов, я решил опросить единственный источник истины - базу данных - и отправлять события SSE всякий раз, когда изменяется состояние. Было бы здорово, если бы мы могли использовать здесь реактивное хранилище данных, но пока я "застрял" с Spring Data JPA и PostgreSQL.
Итак, если у кого-то такая же проблема, я в итоге построил вот так:
@RestController
@RequiredArgsConstructor
class UserController {
private final UserRepository userRepository;
@GetMapping(value = "/{userId}", produces = TEXT_EVENT_STREAM_VALUE)
Flux<ServerSentEvent<ProcessState>> readAsStream(@PathVariable UUID userId) {
return Flux.interval(ZERO, ofSeconds(1L))
.map(n -> userRepository.findById(userId).getProcessState())
.takeUntil(processState -> processState.isTerminalState())
.distinctUntilChanged()
.map(response -> ServerSentEvent.builder((ProcessState) response).build())
.take(ofMinutes(30L));
}
}
На всякий случай, если кому-то интересно: это снова упрощено, чтобы проиллюстрировать проблему. У нас гексагональная архитектура, т.е. не вводимRepositories в нашем @RestControllers, но вызвать бизнес-фасад, также известный как порт ввода, из нашего веб-уровня для получения пользователей.