任何人都可以解释这段代码是如何发送消息给指定的用户的

Question

@Bean
public WebSocketHandler webSocketHandler() {
    TopicProcessor<String> messageProcessor = this.messageProcessor();
    Flux<String> messages = messageProcessor.replay(0).autoConnect();
    Flux<String> outputMessages = Flux.from(messages);

    return (session) -> {
        System.out.println(session);

        session.receive().map(WebSocketMessage::getPayloadAsText).subscribe(messageProcessor::onNext, (e) -> {
            e.printStackTrace();
        });

        return session.getHandshakeInfo().getPrincipal().flatMap((p) -> {
            session.getAttributes().put("username", p.getName());
            return session.send(outputMessages.filter((payload) -> this.filterUser(session, payload))
                    .map((payload) -> this.generateMessage(session, payload)));
        }).switchIfEmpty(Mono.defer(() -> {
            return Mono.error(new BadCredentialsException("Bad Credentials."));
        })).then();
    };
}

私有静态映射客户端=new ConcurrentHashMap（）；public void sendMessageTo（String message,String ToUserName）抛出IOException{

    for (WebSocket item : clients.values()) {
        if (item.username.equals(ToUserName) ) {
            item.session.sendText(message);
            break;
        }
    }
}

我知道所有的消息都存储在outputMessages中并进行子标注。当发出新消息时，它如何找到正确的会话？

Answer 1

我的猜测是WebSocketHandler是一个只包含一个方法handleWebSocketHandler的接口

我相信这又使它成为一个functionalinterface，可以用作lambda。

(session) -> { ... }

因此,当与客户端建立会话时,客户端发送websocket事件。服务器将查找WebSocketHandler并使用来自发送事件的客户端的会话填充它。

class ExampleHandler implements WebSocketHandler {

    @Override
    public Mono<Void> handle(WebSocketSession session) {

            Mono<Void> input = session.receive()
                    .doOnNext(message -> {
                            // ...
                    })
                    .concatMap(message -> {
                            // ...
                    })
                    .then();

            Flux<String> source = ... ;
            Mono<Void> output = session.send(source.map(session::textMessage));

            return Mono.zip(input, output).then();
    }
}

@Bean
public WebSocketHandler webSocketHandler() {
    return new ExampleHandler();
}