Stomp over websocket using Spring and sockJS message lost

Question

On the client side javascript I have

    stomp.subscribe("/topic/path", function (message) {
        console.info("message received");
    });

And on the server side

public class Controller {
  private final MessageSendingOperations<String> messagingTemplate;
  ＠Autowired
  public Controller(MessageSendingOperations<String> messagingTemplate) {
      this.messagingTemplate = messagingTemplate;
  }
  @SubscribeMapping("/topic/path")
  public void subscribe() {
     LOGGER.info("before send");
     messagingTemplate.convertAndSend(/topic/path, "msg");
  }
}

From this setup, I am occasionally (around once in 30 page refreshes) experiencing message dropping, which means I can see neither "message received" msg on the client side nor the websocket traffic from Chrome debugging tool.

"before send" is always logged on the server side.

This looks like that the MessageSendingOperations is not ready when I call it in the subscribe() method. (if I put Thread.sleep(50); before calling messagingTemplate.convertAndSend the problem would disappear (or much less likely to be reproduced))

I wonder if anyone experienced the same before and if there is an event that can tell me MessageSendingOperations is ready or not.

Answer 1

The issue you are facing is laying in the nature of clientInboundChannel which is ExecutorSubscribableChannel by default.

It has 3 subscribers:

0 = {SimpleBrokerMessageHandler@5276} "SimpleBroker[DefaultSubscriptionRegistry[cache[0 destination(s)], registry[0 sessions]]]"
1 = {UserDestinationMessageHandler@5277} "UserDestinationMessageHandler[DefaultUserDestinationResolver[prefix=/user/]]"
2 = {SimpAnnotationMethodMessageHandler@5278} "SimpAnnotationMethodMessageHandler[prefixes=[/app/]]"

which are invoked within taskExecutor, hence asynchronously.

The first one here (SimpleBrokerMessageHandler (or StompBrokerRelayMessageHandler) if you use broker-relay) is responsible to register subscription for the topic.

Your messagingTemplate.convertAndSend(/topic/path, "msg") operation may be performed before the subscription registration for that WebSocket session, because they are performed in the separate threads. Hence the Broker handler doesn't know you to send the message to the session.

The @SubscribeMapping can be configured on method with return, where the result of this method will be sent as a reply to that subscription function on the client.

HTH

Answer 2

Here is my solution. It is along the same lines. Added a ExecutorChannelInterceptor and published a custom SubscriptionSubscribedEvent. The key is to publish the event after the message has been handled by AbstractBrokerMessageHandler which means the subscription has been registered with the broker.

@Override
public void configureClientInboundChannel(ChannelRegistration registration) {
    registration.interceptors(new ExecutorChannelInterceptorAdapter() {

        @Override
        public void afterMessageHandled(Message<?> message, MessageChannel channel, MessageHandler handler, Exception ex) {
            SimpMessageHeaderAccessor accessor = SimpMessageHeaderAccessor.wrap(message);
            if (accessor.getMessageType() == SimpMessageType.SUBSCRIBE && handler instanceof AbstractBrokerMessageHandler) {
                /*
                 * Publish a new session subscribed event AFTER the client
                 * has been subscribed to the broker. Before spring was
                 * publishing the event after receiving the message but not
                 * necessarily after the subscription occurred. There was a
                 * race condition because the subscription was being done on
                 * a separate thread.
                 */
                applicationEventPublisher.publishEvent(new SessionSubscribedEvent(this, message));
            }
        }
    });

}