Spring WebFlux： WebSocketSession.send（）不发送消息

Question

Helloc我试图使用Spring WebFlux创建webscoket终结点。我希望这个endpoint返回一些事件。
为了做到这一点，我创建了事件的ConnectableFlux，并在句柄（...）方法中将其映射到Flux。但是在我把它给WebSocket会话之后，什么都没有发生——webSocket会话客户端没有收到任何东西。但是同时println（event.toString（）），你可以在下面的句柄（...）方法中看到它实际上将信息打印到控制台。
你能告诉我错过了什么吗？

public class EventWebsocketHandler implements WebSocketHandler {

    //  constructors and etc.

    @Override
    public Mono<Void> handle(WebSocketSession session) {
        ObjectMapper objectMapper = new ObjectMapper();

        Flux<WebSocketMessage> messages = eventService.events()
                .flatMap(event -> {
                    try {
                        System.out.println(event.toString());
                        return Mono.just(objectMapper.writeValueAsString(event));
                    } catch (JsonProcessingException e) {
                        return Mono.error(e);
                    }
                })
                .map(session::textMessage);

        return session.send(messages);
    }

@Service
public class EventService {

    List<EventDto> events = new ArrayList<>();

    private final Flux<EventDto> eventFlux = Flux.<EventDto>create(fluxSink -> {
        while (true) {
            if (!events.isEmpty()) {
                fluxSink.next(events.get(0));
                events.remove(0);
            }
        }
    })
            .publish()
            .autoConnect();

    public void push(EventDto event) {
        events.add(event);
    }

    public Flux<EventDto> events() {
        return eventFlux;
    }

}

我在我的项目中有另一个WebSocketHandler，它工作正常，这意味着配置一切正常：

public class MyWebSocketHandler implements WebSocketHandler {

    @Override
    public Mono<Void> handle(WebSocketSession session) {
        Flux<Long> source = Flux.interval(Duration.ofMillis(1000 * 3));
        return session.send(source.map(l -> session.textMessage(String.valueOf(l))));
    }
}

Answer 1

这

    private final Flux<EventDto> eventFlux = Flux.<EventDto>create(fluxSink -> {
    while (true) {
        if (!events.isEmpty()) {
            fluxSink.next(events.get(0));
            events.remove(0);
        }
    }
})
        .publish()
        .autoConnect();

必须用这个来代替

private final Sinks.Many<EventDto> processor = Sinks.many().multicast().onBackpressureBuffer();