如何将消息广播到ConnectableFlux中的订户

Question

我是Spring WebFlux的新手，我在此链接中使用Spring WebFlux和Websocket找到了此聊天演示：

https://blog.monkey.codes/how-to-build-a-chat-app-using-webflux-websockets-react/

根据文章，它说：

该应用程序的关键是将WebSocketSessions连接到一个另一个。这是通过连接的传入消息流来实现的每次与全球发布商联系。另一方面，每个会话订阅全球发布者产生的消息。

反应器配置：

    @Bean
    public UnicastProcessor<Event> eventPublisher(){
        return UnicastProcessor.create();
    }

    @Bean
    public Flux<Event> events(UnicastProcessor<Event> eventPublisher) {
        return eventPublisher
                .replay(25)
                .autoConnect();
    }

    @Bean
    public HandlerMapping webSocketMapping(UnicastProcessor<Event> eventPublisher, Flux<Event> events) {
        Map<String, Object> map = new HashMap<>();
        map.put("/websocket/chat", new ChatSocketHandler(eventPublisher, events));
        SimpleUrlHandlerMapping simpleUrlHandlerMapping = new SimpleUrlHandlerMapping();
        simpleUrlHandlerMapping.setUrlMap(map);

        //Without the order things break :-/
        simpleUrlHandlerMapping.setOrder(10);
        return simpleUrlHandlerMapping;
    }

    @Bean
    public WebSocketHandlerAdapter handlerAdapter() {
        return new WebSocketHandlerAdapter();
    }

聊天套接字处理程序：

@Override
public Mono<Void> handle(WebSocketSession session) {
    WebSocketMessageSubscriber subscriber = new WebSocketMessageSubscriber(eventPublisher);
    session.receive()
            .map(WebSocketMessage::getPayloadAsText)
            .map(this::toEvent)
            .subscribe(subscriber::onNext, subscriber::onError, subscriber::onComplete);
    return session.send(outputEvents.map(session::textMessage));
}

private static class WebSocketMessageSubscriber {
    private UnicastProcessor<Event> eventPublisher;
    private Optional<Event> lastReceivedEvent = Optional.empty();

    public WebSocketMessageSubscriber(UnicastProcessor<Event> eventPublisher) {
        this.eventPublisher = eventPublisher;
    }

    public void onNext(Event event) {
        lastReceivedEvent = Optional.of(event);
        eventPublisher.onNext(event);
    }

    public void onError(Throwable error) {
        //TODO log error
        error.printStackTrace();
    }

    public void onComplete() {
        lastReceivedEvent.ifPresent(event -> eventPublisher.onNext(
                Event.type(USER_LEFT)
                        .withPayload()
                        .user(event.getUser())
                        .build()));
    }

}

[如果一个用户发送了一条消息，则其他用户可以看到此消息，因为它已广播给所有用户。但是，当我阅读代码时，我不明白ConnectableFlux（事件）如何将消息发送到所有Web套接字会话。我也尝试注释UserStats类的代码，但它仍然可以正常工作。谁能向我解释一下？

Answer 1

通过Websocket从服务器向浏览器发送消息的代码是：

session.send(outputEvents.map(session::textMessage))

换句话说，每当事件（消息）被推送到outputEvents通量时，它将通过链接到会话的Websocket发送到浏览器。

之所以像广播一样，是因为每个为每个新WebSocketSession创建的WebSocketMessageSubscriber都会将接收到的所有内容（从浏览器到服务器）发布到eventPublisher上，从而使它在outputEvents Flux上可见。

  public void onNext(Event event) {
    lastReceivedEvent = Optional.of(event);
    eventPublisher.onNext(event);
  }

outputEvents通量在所有WebSocketMessageSubscribers中共享，并且通过此通量接收的所有事件都通过Websocket发送回浏览器，包括从该Websocket接收到的事件。

在下面的“全局消息流”中的图片表示outputEvents通量。