Блог

У меня есть приложение для чата с spring reactive webflux. Когда создается новое сообщение, все подписчики получают это сообщение (это просто пример для упрощения моей проблемы). Все правильно и работало безукоризненно. Но теперь мне нужно событие для подписчиков при поступлении нового сообщения.

Это мой код:

Контроллер:

 @Autowired
@Qualifier("ptpReplayProcessor")
private ReplayProcessor<String> ptpReplayProcessor;

@GetMapping(value = "/chat/subscribe", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> subscribe() {
    return Flux.from(ptpReplayProcessor);
}
  

ReplayProcessorConfig:

 @Configuration
public class ReplayProcessorConfig {
    @Bean("ptpReplayProcessor")
    ReplayProcessor<String> ptpReplayProcessor() {
        ReplayProcessor<String> replayProcessor = ReplayProcessor.create(0, false);
        replayProcessor.subscribe(new BaseSubscriber<String>() {
            @Override
            protected void hookOnNext(String ptp) {
                System.out.println("replay processor called!");
            }

        });
        return replayProcessor;
    }
}
  

pom.xml

 <dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-core</artifactId>
</dependency>
  

Когда создается новое сообщение, я вызываю ptpReplayProcessor.onNext(message) . Это работает правильно, и все клиенты получают сообщение, но фраза replay processor called! просто выводится от отправителя сообщения. Я хочу вызвать событие для всех клиентов при получении нового сообщения. Также я пробовал методы ReplayProcessor.doOnNext() и ReplayProcessor.doOnEach() , но не сработало. Есть ли какой-либо способ сделать это?

вы можете сделать это в своем методе подписки:

 return Flux.from(ptpReplayProcessor).doOnNext(s -> System.out.println("new message has been sent"));