6

With the Stomp broker relay for web socket messaging, I can subscribe to a destination /topic/mydest. This creates a broker subscription and receives all messages that something in the system triggers for this broker destination, which happens when some event in the system occurs.

I can subscribe to a destination /app/mydest, and a controller method with @SubscribeMapping("mydest") will be called, and the return value is sent back only on this socket as a message. As far as I can tell, this is the only message that will ever be sent for this subscription.

Is there a way to combine this in a single subscription, i.e. create a broker subscription for a certain /topic destination, and trigger some code that directly sends a message back to the subscriber?

The use case: when an error occurs in the system, a message with a current error count is sent to /topic/mydest. When a new client subscribes, I want to send him only the last known error count. Others are not interested at this moment, as the count has not changed.

My current solution is to subscribe to both /app/mydest and /topic/mydest and use the same message handler on the client. But it really is one logical subscription, and it is a bit error prone as a client needs to remember to subscribe to both.

My questions in this context: will there ever be a further message for the /app/ subscription? Is there anything to call to trigger one? How else can I send initial information to a subscriber for a topic, without sending redundant messages to the existing subscribers?

As requested, here's my Websocket configuration class.

@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"); } }
Improve this question
1
  • Perhaps I should clarify: the current solution that I describe is fine enough from a practical standpoint. My main motivation for the question is to fully grasp the intent of subscriptions to user destinations, and ways to act on subscriptions to broker destinations. Commented Feb 9, 2017 at 19:33

3 Answers 3

Reset to default
0

You can use ApplicationListener and SessionSubscribeEvent. Example:

@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"); } }
Improve this answer
Sign up to request clarification or add additional context in comments.

6 Comments

I'll try that. But doesn't that need an additional subscription to /user/topic/mydest?
No, just subscription to /topic/mydest. Tested with my app - working fine
It does not work for me. The idea of using the SessionSubscribeEvent to act on is a valuable hint, and I appreciate it. But sending a message to the user destination /topic/mydest is not received on the new subscription to the regular broker destination /topic/mydest. Frankly, I don't see any hint anywhere in the docs to support that assumption. Of course, I can do messagingTemplate.convertAndSend("/topic/mydest") in the event listener and send a message to every subscriber. While that wouldn't really hurt for that kind of message, it's not what I want to do.
@rainerfrey could you please update the question and provide your WebSocket configuration?
If this works for anyone please send the github link, please.
|
0

I think I found a solution for your problem:

You have to subscribe to the user specific topic. In my example I've created a topic /topic/progress.

I subscribe to /user/topic/progress

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

I've created a component that listens for the SessionSubscribeEvent in order to react on new subscriptions:

@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()); } } }

The WebSocketservice is used to send the messages to the subscribing users. I have a method to broadcast stuff to a topic and one to send it to specific users only. The broadcast method isn't the original one. I use the SimpUserRegistry to get all subscribers and send the message to each separately:

@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); } }
Improve this answer

Comments

-1

You can listen to the session subscribe event and send initial message

@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"); } } }

}

Improve this answer

Comments

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.