Broken WebSocket subscription using simple message broker

[BenS89]

We used WebSockets within Spring Boot 2.3.5 using a simple message broker. After the update to Spring Boot 2.4 / Spring Framework 5.3, we noticed that our web socket mechanism is not working anymore. We use the following web socket config:

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    public void configureMessageBroker(MessageBrokerRegistry config) {
        config.enableSimpleBroker("/topic", "/user");
        config.setApplicationDestinationPrefixes("/app");
        config.setUserDestinationPrefix("/user");
    }

    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/websocket");
        registry.addEndpoint("/websocket").setAllowedOrigins("http://localhost:8080").withSockJS();
    }

    @Bean(name = "websocketLocaleResolver")
    public SessionLocaleResolver websocketLocaleResolver() {
        SessionLocaleResolver websocketLocaleResolver = new SessionLocaleResolver();
        websocketLocaleResolver.setDefaultLocale(Locale.GERMANY);

        return websocketLocaleResolver;
    }

}

A look at the DefaultSubscriptionRegistry shows that there changed a lot comparing to Spring Boot 2.3.5. We could nail down the issue because we saw that even though our web socket subscription was successful within the DefaultSubscriptionRegistry, the method this.subscriptionRegistry.findSubscriptions(message) within the SimpleBrokerMessageHandler always returns an empty map (in contrast to Spring Boot 2.3).

protected void sendMessageToSubscribers(@Nullable String destination, Message<?> message) {
		MultiValueMap<String,String> subscriptions = this.subscriptionRegistry.findSubscriptions(message);
...

When rolling back to Spring 2.3.5, the same web socket business logic works fine with the old implementations of DefaultSubscriptionRegistry and SimpleBrokerMessageHandler.

We also noticed that with Spring Boot 2.3.5 the DestinationCache returns a map including a SessionId and SubscriptionId (expected behavior). Using Spring Boot 2.4.0, the returned LinkedMultiValueMap sessionIdToSubscriptionIds is always empty.

public LinkedMultiValueMap<String, String> getSubscriptions(String destination) {
			LinkedMultiValueMap<String, String> sessionIdToSubscriptionIds = this.destinationCache.get(destination);
			if (sessionIdToSubscriptionIds == null) {
				sessionIdToSubscriptionIds = this.destinationCache.computeIfAbsent(destination, _destination -> {
					LinkedMultiValueMap<String, String> matches = computeMatchingSubscriptions(destination);
					// Update queue first, so that cacheSize <= queue.size(
					this.cacheEvictionPolicy.add(destination);
					this.cacheSize.incrementAndGet();
					return matches;
				});
				ensureCacheLimit();
			}
			return sessionIdToSubscriptionIds;
		}

[snicoll]

Thanks for the report but none of that code is part of Spring Boot. I am transferring the issue to the Spring Framework issue tracker so that the team can have a look.

[rstoyanchev]

Yes there was a major refactoring in DefaultSubscriptionRegistry to reduce locking contention. Is this something that you can reproduce reliably, and if yes you can provide more detailed instructions or a sample?

[BenS89]

Thanks for your feedback. In the meantime, we found the root cause of the issue. In Spring Framework 5.3, only one destination per {sessionId, subscriptionId} is allowed within the DefaultSubscriptionRegistry:

Spring-5.3.1 -> DefaultSubscriptionRegistry$SessionRegistry
    public void addSubscription(String sessionId, Subscription subscription) {
        SessionInfo info = this.sessions.computeIfAbsent(sessionId, _sessionId -> new SessionInfo());
        info.addSubscription(subscription);

Spring-5.3.1 -> DefaultSubscriptionRegistry$SessionInfo
    public void addSubscription(Subscription subscription) {
        this.subscriptionMap.putIfAbsent(subscription.getId(), subscription);

In Spring 5.2, storing multiple destinations for this tuple is possible:

Spring-5.2.10 -> DefaultSubscriptionRegistry$SessionSubscriptionRegistry
    public SessionSubscriptionInfo addSubscription(String sessionId, String subscriptionId,
            String destination, @Nullable Expression selectorExpression) {
        SessionSubscriptionInfo info = this.sessions.get(sessionId);
        if (info == null) {
            info = new SessionSubscriptionInfo(sessionId);
            SessionSubscriptionInfo value = this.sessions.putIfAbsent(sessionId, info);
            if (value != null) {
                info = value;
            }
        }
        info.addSubscription(destination, subscriptionId, selectorExpression);
        return info;

Spring-5.2.10 -> DefaultSubscriptionRegistry$SessionSubscriptionInfo
    public void addSubscription(String destination, String subscriptionId, @Nullable Expression selectorExpression) {
        Set<Subscription> subs = this.destinationLookup.get(destination);
        if (subs == null) {
            synchronized (this.destinationLookup) {
                subs = this.destinationLookup.get(destination);
                if (subs == null) {
                    subs = new CopyOnWriteArraySet<>();
                    this.destinationLookup.put(destination, subs);
                }
            }
        }
        subs.add(new Subscription(subscriptionId, selectorExpression));

In our frontend, we use the following code for the web socket subscription:

public subscribe(callback: any, locale: string): void {

        this.subscribers.push(callback);

        const url = '/websocket';

        const socket = new SockJS(url);
        this.client = Stomp.over(socket);

        this.client.debug = () => { };

        const headers = {
            locale: locale
        };

        this.client.connect(headers, frame => {
            this.client.subscribe('/user/topic/unread', message => this.onMessage(message));
            this.client.send('/app/unread', headers, "{}");
        });

This results in two requests having different destinations with the same session ID / subscription ID, for example:

Request 1:

Request 2:

For the second request, we can not save the destination because in addSubscription this.subscriptionMap already has an entry for the given subscription ID:

public void addSubscription(Subscription subscription) {
	this.subscriptionMap.putIfAbsent(subscription.getId(), subscription);
}

Therefore, the main difference between Spring 5.2 and 5.3 seems like the ability to store multiple destinations for a given {sessionId, subscriptionId} within the old 5.2 implementation.

Was it intended to remove the ability to store multiple destinations for existing session/sub ids?

[rstoyanchev]

@BenS89 thanks for the extra details.

In the STOMP protocol the same user can subscribe more than once to the same destination but each subscription must have a unique id. You can also see it on the MESSAGE frame (from server to client) which includes only the subscription id, and not the destination and that is how messages can be correlated to the subscription. So you can subscribe twice to the same destination and still get messages for each subscription separately since each subscription should have a unique id.

I don't quite understand how you make use of this but it is unusual for broker and user destinations to overlap, and I would not expect to see that. A user prefixed destination is just a general representation of a topic (let's category) that is meant to be transformed into a unique destination specific to the user. So the broker should only get one subscription for the user.

[BenS89]

Thanks for your feedback. If I unterstand you right, the mentioned change was intended. We found a way to avoid one of the two requests. That fixed the issue for us as well. Therefore, the issue can be closed.