Spring web socket 消息传递 - 订阅和发送初始消息

Question

使用用于 Web 套接字消息传递的 Stomp 代理中继，我可以订阅目标 /topic/mydest。这将创建一个代理订阅，并接收系统中的某些东西为此代理目标触发的所有消息，这会在系统中的某些事件发生时发生。

我可以订阅一个目的地/app/mydest，一个带有@SubscribeMapping("mydest") 的控制器方法将被调用，返回值只在这个套接字上作为消息发送回来。据我所知，这是唯一会为此订阅发送的消息。

有没有办法将这些合并到一个订阅中，即为某个/topic 目的地创建一个代理订阅，并触发一些代码，直接将消息发送回订阅者？

用例：当系统发生错误时，将带有当前错误计数的消息发送到/topic/mydest。当一个新客户订阅时，我只想向他发送最后一个已知的错误计数。其他人暂时不感兴趣，因为计数没有改变。

我当前的解决方案是同时订阅/app/mydest 和/topic/mydest 并在客户端上使用相同的消息处理程序。但它确实是一个合乎逻辑的订阅，而且有点容易出错，因为客户需要记住同时订阅两者。

在这种情况下我的问题是：/app/ 订阅是否还会有更多消息？有什么可以触发的吗？在不向现有订阅者发送冗余消息的情况下，我还能如何将初始信息发送给某个主题的订阅者？

根据要求，这是我的 Websocket 配置类。

@Configuration
@EnableWebSocketMessageBroker
public class WebsocketConfiguration extends AbstractWebSocketMessageBrokerConfigurer {
    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/ws").setAllowedOrigins("*").withSockJS();
    }

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.enableStompBrokerRelay("/queue/", "/topic/", "/exchange/");
        registry.setApplicationDestinationPrefixes("/app");
    }
}

Answer 1

您可以使用ApplicationListener 和SessionSubscribeEvent。 示例：

@Component
public class SubscribeListener implements ApplicationListener<SessionSubscribeEvent> {

    private final SimpMessagingTemplate messagingTemplate;

    @Autowired
    public SubscribeListener(SimpMessagingTemplate messagingTemplate) {
        this.messagingTemplate = messagingTemplate;
    }

    @Override
    public void onApplicationEvent(SessionSubscribeEvent event) {
        messagingTemplate.convertAndSendToUser(event.getUser().getName(), "/topic/mydest", "Last known error count");
    }
}

Answer 2

您可以监听会话订阅事件并发送初始消息

@Component
@RequiredArgsConstructor
public class WebSocketEventListener {

private static final Logger logger = LoggerFactory.getLogger(WebSocketEventListener.class);
private final SimpMessagingTemplate simpMessagingTemplate;


@EventListener
public void handleSessionSubscribeEvent(SessionSubscribeEvent event) {
    logger.info("Subscribed to session: " + event);
    Principal user = event.getUser();
    if (user instanceof UsernamePasswordAuthenticationToken) {
        UsernamePasswordAuthenticationToken token = (UsernamePasswordAuthenticationToken) user;
        if (token.getPrincipal() instanceof UserDetails) {
            UserDetails userDetails = (UserDetails) token.getPrincipal();
            simpMessagingTemplate.convertAndSendToUser(userDetails.getUsername(), "/queue/notify", "Hello");
        }
    }
}

}

Answer 3

我想我找到了解决您问题的方法：

您必须订阅用户特定的主题。 在我的示例中，我创建了一个主题 /topic/progress。

我订阅/user/topic/progress

stompClient.subscribe('/user/topic/progress', progressMessage => {
      ...
})

我创建了一个组件来侦听SessionSubscribeEvent，以便对新订阅做出反应：

@Component
public class WebSocketEventListener {

    @Autowired
    private WebSocketService webSocketService;

    @Autowired
    private ProgressService progressService;

    @EventListener
    public void handleWebSocketConnectListener(SessionSubscribeEvent event) throws IllegalAccessException {
        if(Objects.equals(event.getMessage().getHeaders().get("simpDestination"), "/user/topic/progress")) {
            webSocketService.sendCurrentProgessToUser(progressService.getProgress(), event.getUser().getName());
        }
    }
}

WebSocket 服务用于向订阅用户发送消息。 我有一种方法可以将内容广播到某个主题，并且可以仅将其发送给特定用户。广播方式不是原来的方式。我使用SimpUserRegistry 来获取所有订阅者并将消息分别发送给每个订阅者：

@Controller
@Service
public class WebSocketService {

    @Autowired
    private SimpMessagingTemplate simpMessagingTemplate;

    @Autowired
    private SimpUserRegistry simpUserRegistry;

    private static final String WS_PROGRESS_DESTINATION = "/topic/progress";

    public void broadcastCurrentProgress(Progress progress) {
        ProgressDto progressDto = new ProgressDto(progress);
        List<String> subscribers = simpUserRegistry.getUsers().stream()
                .map(SimpUser::getName).collect(Collectors.toList());
        for(String username : subscribers) {
            simpMessagingTemplate.convertAndSendToUser(username, WS_PROGRESS_DESTINATION, progressDto);
        }
    }

    public void sendCurrentProgessToUser(Progress progress, String name) {
        ProgressDto progressDto = new ProgressDto(progress);
        simpMessagingTemplate.convertAndSendToUser(name, WS_PROGRESS_DESTINATION, progressDto);
    }
}