Allow multiple destinations per subscription in DefaultSubscriptionRegistry

[alienisty]

Issue #24395 broke "user" destinations because it made subscription-id->destination a one to one association whereas it is a one to many.

This restores such state while keeping the optimisations done in #24395.

[pivotal-cla]

@alienisty Please sign the Contributor License Agreement!

Click here to manually synchronize the status of this Pull Request.

See the FAQ for frequently asked questions.

[pivotal-cla]

@alienisty Thank you for signing the Contributor License Agreement!

[rstoyanchev]

Subscription id's must be unique. From the STOMP spec:

Since a single connection can have multiple open subscriptions with a server, an id header MUST be included in the frame to uniquely identify the subscription.

This has been discussed at length under #26118 and #26474. I believe this is likely the same issue.

Can you share what your @EnableWebSocketMessageBroker config looks like? In particular what, if any, destination prefixes, are configured.

[alienisty]

Subscription id's must be unique. From the STOMP spec:

Since a single connection can have multiple open subscriptions with a server, an id header MUST be included in the frame to uniquely identify the subscription.

This has been discussed at length under #26118 and #26474. I believe this is likely the same issue.

Can you share what your @EnableWebSocketMessageBroker config looks like? In particular what, if any, destination prefixes, are configured.

We don't use the annotation, we extend DelegatingWebSocketMessageBrokerConfiguration directly:

@Configuration
public static class MessageBrokerConfiguration
    extends DelegatingWebSocketMessageBrokerConfiguration
    implements ApplicationListener<ApplicationPreparedEvent> {

  private final ApplicationProperties.WebSocket.StompBrokerRelay config;

  private final WebSocketMessageBrokerStats webSocketMessageBrokerStats;

  public MessageBrokerConfiguration(
      ApplicationContext context,
      ApplicationProperties properties,
      @Lazy WebSocketMessageBrokerStats webSocketMessageBrokerStats) {
    // We need to ensure that the application context is available when the STOMP sub-protocol bean is built,
    // otherwise the session events won't work. (see WebSocketMessageBrokerConfigurationSupport.stompWebSocketHandlerMapping())
    this.setApplicationContext(context);
    this.config = properties.getWebSocket().getStompBrokerRelay();
    this.webSocketMessageBrokerStats = webSocketMessageBrokerStats;
  }

  @Override
  public void onApplicationEvent(ApplicationPreparedEvent event) {
    long statsLoggingPeriod = config.getStatsLoggingPeriod();
    webSocketMessageBrokerStats.setLoggingPeriod(MINUTES.toMillis(statsLoggingPeriod));
  }
}

As you can see there is nothing changing how the protocol is implemented and the superclass creates and registers an instance of SubProtocolWebSocketHandler and UserDestinationMessageHandler into the ExecutorSubscribableChannel. In ExecutorSubscribableChannel.sendInternal(), the same Message, therefore with the same subscription-id header is passed to both handlers. When the message is a "SUBSCRIBE", both hanlders will try to subscribe their destinations with the same subscription-id.

I can appreciate what the STOMP protocol specification says, I would argue, though, that the subscription-id must be unique per connection not per destination, so much so that messages need to specify both subscription-id and destination; if they were a one-to-one association, the destination header would be redundant.
We are also using the latest StompJs library which also claims to be a rewrite which is more adherent to the STOMP spec and it works just fine receiving messages to different destinations but same subscription-id.

At the very least we have a problem with how user destinations are registered.

[rstoyanchev]

I think your issue is exactly the same.

If there aren't any destination prefixes configured for messages to the broker, then the broker ends up having all messages forwarded to it, but "/user" prefixed messages aren't meant to go directly there. Instead, UserDestinationMessageHandler is the only thing that should handle such messages. It transforms the destination to a "/queue" prefixed, unique destination, and that's what's then sent to the broker. This is described in the reference. In your case however the broker sees both the "/user/" and the transformed "/queue" prefixed messages with the same id, but it should see only the latter.

Can you try configuring prefixes for the messages that should go to the broker? See the discussion and recommendations under #26474.

[alienisty]

Hi @rstoyanchev, setting the user destination prefix to "/user" fixes the problem, but I think that either the reference documentation should be changed or such prefix should be "/user" by default. In fact, the reference says:

An application can send messages that target a specific user, and Spring’s STOMP support recognizes destinations prefixed with /user/ for this purpose. For example, a client might subscribe to the /user/queue/position-updates destination. This destination is handled by the UserDestinationMessageHandler and transformed into a destination unique to the user session.

which is true, but the reference documentation doesn't mention the user destination prefix, and for sending a message using a messaging template, it only says:

You can send a message to user destinations from any application component by, for example, injecting the SimpMessagingTemplate created by the Java configuration or the XML namespace. (The bean name is brokerMessagingTemplate if required for qualification with @qualifier.) The following example shows how to do so:

@Service
public class TradeServiceImpl implements TradeService {

    private final SimpMessagingTemplate messagingTemplate;

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

    // ...

    public void afterTradeExecuted(Trade trade) {
        this.messagingTemplate.convertAndSendToUser(
                trade.getUserName(), "/queue/position-updates", trade.getResult());
    }
}

which now it does not work unless we set the user destination prefix to "/user".

[rstoyanchev]

I have added a3655c4 to add some guidance to the reference docs as suggested and will close this issue.

For 6.0, we could consider having the broker explicitly filter out messages with user destinations when it is itself not configured with any specific prefixes, but that would be better as a separate issue.