为什么我可以订阅目的地，但我没有收到任何消息为该用户与Spring WebSocket STOMP（与Spring Security性和KeyCloak）？

Question

我试图实现通过WebSocket发送消息的微服务。

我可以将正确的消息发送到子集和身份验证的客户端（将JWT令牌传递给WebClient服务器），但现在我想只向特定用户发送消息。

根据spring官方文档，我可以正确订阅客户端，但没有收到任何消息。

WebSocket服务器配置：

• WebSecurityConfig.java

[...]
    @Override
    protected void configure(HttpSecurity http) throws Exception {
        super.configure(http);
        http.cors()
                .and().csrf().disable()

                .authorizeRequests()
                .anyRequest()
                .authenticated()
        ;
    }

    @Override
    public void configure(WebSecurity web) {
        web.ignoring()
                .antMatchers("/ws/**");
    }
[...]

• WebSocketConfig.java

@Slf4j
@Configuration
//@EnableScheduling
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    @Qualifier("websocket")
    private AuthenticationManager authenticationManager;

    @Autowired
    public WebSocketConfig(AuthenticationManager authenticationManager)
    {
        this.authenticationManager = authenticationManager;
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/ws").setAllowedOrigins("*")
        registry.addEndpoint("/ws").setAllowedOrigins("*").withSockJS();
    }

    @Override
    public void configureClientInboundChannel(ChannelRegistration registration) {
        registration.interceptors(new ChannelInterceptor() {
            @Override
            public Message<?> preSend(Message<?> message, MessageChannel channel) {
                StompHeaderAccessor headerAccessor =
                        MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);

                if (StompCommand.CONNECT.equals(headerAccessor.getCommand()))
                {
//                  log.info("TOKEN: {}", headerAccessor.getNativeHeader("token").toString());
                    Optional.ofNullable(headerAccessor.getNativeHeader("token")).ifPresent(ah ->
                    {
                        String bearerToken = ah.get(0).replace("Bearer ", "");
                        JWSAuthenticationToken token = (JWSAuthenticationToken) authenticationManager
                                .authenticate(new JWSAuthenticationToken(bearerToken));
                        headerAccessor.setUser(token);
                    });
                }

                return message;
            }
        });
    }

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {

        registry.setApplicationDestinationPrefixes("/app");
        registry.enableSimpleBroker("/topic", "/queue");
    }
}

• WebSocketSecurityConfig。爪哇

@Configuration
public class WebSocketSecurityConfig extends AbstractSecurityWebSocketMessageBrokerConfigurer {

    @Override
    protected void configureInbound(MessageSecurityMetadataSourceRegistry messages) {

        messages
                .simpTypeMatchers(CONNECT, HEARTBEAT, UNSUBSCRIBE, DISCONNECT).permitAll()
                .simpDestMatchers("/app/**", "/topic/**", "/queue/**").authenticated()
                .simpSubscribeDestMatchers("/topic/**", "/queue/**", "/user/**").authenticated()
                .anyMessage().denyAll();
    }

    @Override
    protected boolean sameOriginDisabled() {
        //disable CSRF for websockets for now...
        return true;
    }
}

• 在一个Java的类中，我尝试了两个用于发送消息的函数：

simpMessagingTemplate.convertAndSend("/user/" +  username + "/queue/private-messages", message);


simpMessagingTemplate.convertAndSendToUser("user","/queue/private-messages", message);

WebSocket客户端配置

• WebSecurityConfig.java

[...]
    @Override
    protected void configure(HttpSecurity http) throws Exception {

        super.configure(http);

        http.cors()
                .and().csrf().disable()

                .authorizeRequests()
                .anyRequest()
                .authenticated()

        ;
    }
[...]

• StompClient。java，其中我使用函数createAndConnectClient创建连接

[...]
    public void createAndConnectClient(String accessToken) {

        WebSocketClient client = new StandardWebSocketClient();
        WebSocketStompClient stompClient = new WebSocketStompClient(client);

        stompClient.setMessageConverter(new MappingJackson2MessageConverter());

        StompSessionHandler sessionHandler = new MyStompSessionHandler(stompConfig);

        // connect with custom headers
        final WebSocketHttpHeaders headers = new WebSocketHttpHeaders();
        final StompHeaders head = new StompHeaders();
        head.add("token", accessToken);
        stompClient.connect(serverURL, headers, head, sessionHandler);
    }
[...]

• MyStompSessionHandler

@Component
public class MyStompSessionHandler extends StompSessionHandlerAdapter {

    private final StompConfig stompConfig;

    @Autowired
    public MyStompSessionHandler(StompConfig stompConfig) {
        this.stompConfig = stompConfig;
    }


    @Override
    public void afterConnected(StompSession session, StompHeaders connectedHeaders) {
        log.info("New session established : " + session.getSessionId());
        session.subscribe("/user/queue/private-messages", this);
        log.info("Subscribed to /user/queue/private-messages");
    }

    @Override
    public void handleException(StompSession session, StompCommand command, StompHeaders headers, byte[] payload, Throwable exception) {
        log.error("Got an exception", exception);
    }

    @Override
    public Type getPayloadType(StompHeaders headers) {
        return Message.class;
    }

    @Override
    public void handleFrame(StompHeaders headers, Object payload) {

        if (payload != null) {
            Message msg = (Message) payload;
            log.info("Received : " + msg.getText() + " from : " + msg.getFrom());
        } else {
            log.info("NULL Payload");
        }
    }
}

正如我所说，我可以订阅目的地，但我没有收到任何消息。

编辑：我在发送消息时添加了此代码

[...]
SimpMessageHeaderAccessor headerAccessor = SimpMessageHeaderAccessor.create(SimpMessageType.MESSAGE);
JWSAuthenticationToken token = null;
    try {
            token = userService.getToken(message.getTo());
            log.info("TOKEN BEFORE SEND: {}", token);
        } catch (NotFoundException e) {
            e.printStackTrace();
            log.error("User isn't connected");
        }
        headerAccessor.setUser(token);
        headerAccessor.setSessionId(userService.getSessionId(message.getTo()));
        log.info("SESSION-ID: {}", headerAccessor.getSessionId());
        log.info("HEADER: {}", headerAccessor.toString());
        simpMessagingTemplate.convertAndSendToUser(message.getTo(), webSocketConfig.getDestination(), message, headerAccessor.getMessageHeaders());
[...]

所以。。。我确信用户（以及SessionID）是相同的。但仍然没有收到任何关于我客户的信息！

工作解决方案：

不是我想要的东西但是。。。它起作用了！¯_(ツ)_/¯

（没有受益于STOMP用户管理机制）

当我订阅目的地时，在我的客户端中，我将用户的名称添加到detionation url

“/queue/messages-”用户。getName（）。

使用该目的地上的convertAndSend（发送消息且不带标题的服务器端），我可以将私人消息发送给用户。

Answer 1

不是我想要的东西但是。。。它起作用了！¯_(ツ)_/¯

（没有受益于STOMP用户管理机制）

当我订阅目的地，在我的客户端，我添加到destionation url的用户名：

“/queue/messages-”用户。getName（）。

使用该目的地（发送消息的服务器端）上的convertAndSend，我可以向用户发送私人消息。

Answer 2

是否为SimpMessageTemplate设置了UserDestinationPrefix。如果没有，请尝试将其设置为“/user”

Answer 3

我相信发生这种情况是因为DefaultSimpUser注册表没有与用户一起更新。如果您看到这一行，它应该添加用户，它不是从STOMP头erAccesssor添加，而是从子协议（即web套接字协议）添加。因此，您必须为webSocket会话设置用户。

private void populateSecurityContextHolder(String
            username, List<SimpleGrantedAuthority> grantedAuthorities) {
        PreAuthenticatedAuthenticationToken authToken = new PreAuthenticatedAuthenticationToken(
                username, null, grantedAuthorities);
        SecurityContextHolder.getContext().setAuthentication(authToken);
    }

因此，设置此选项后，用户注册表应该随用户一起更新，因此当消息发送到队列时，当它找到用户时，应该能够将消息发送给用户，而不是因为用户不在场而丢弃消息。