Ensure matching user destination returned

Before this change, when a client subscribed to a "user" destination
(e.g. /user/foo), actual messages received in response to that
subscription contained the server-translated, unique user destination
(e.g. /foo-user123).

This is not an issue for clients such as stomp.js since the
subscription is unique and sufficient to match subscription responses.
However, other STOMP clients do additional checks on the destination
of the subscription and the response.

This change ensures that messages sent to clients on user destionations
always contain a destination that matches the one on the original
subscription.

Issue: SPR-11423

Author: rstoyanchev

spring-messaging/src/main/java/org/springframework/messaging/simp/user/DefaultUserDestinationResolver.java

@@ -30,10 +30,21 @@
 import java.util.Set;
 
 /**
- * A default implementation of {@link UserDestinationResolver}.
+ * A default implementation of {@link UserDestinationResolver} that relies
+ * on the {@link org.springframework.messaging.simp.user.UserSessionRegistry}
+ * provided to the constructor to find the sessionIds associated with a user
+ * and then uses the sessionId to make the target destination unique.
  * <p>
- * Uses the {@link org.springframework.messaging.simp.user.UserSessionRegistry}
- * provided to the constructor to find the sessionIds associated with a user.
+ * When a user attempts to subscribe to "/user/queue/position-updates", the
+ * "/user" prefix is removed and a unique suffix added, resulting in something
+ * like "/queue/position-updates-useri9oqdfzo" where the suffix is based on the
+ * user's session and ensures it does not collide with any other users attempting
+ * to subscribe to "/user/queue/position-updates".
+ * <p>
+ * When a message is sent to a user with a destination such as
+ * "/user/{username}/queue/position-updates", the "/user/{username}" prefix is
+ * removed and the suffix added, resulting in something like
+ * "/queue/position-updates-useri9oqdfzo".
  *
  * @author Rossen Stoyanchev
  * @since 4.0
@@ -87,30 +98,32 @@ public UserSessionRegistry getUserSessionRegistry() {
 	}
 
 	@Override
-	public Set<String> resolveDestination(Message<?> message) {
+	public UserDestinationResult resolveDestination(Message<?> message) {
 
 		SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
-		UserDestinationInfo info = getUserDestinationInfo(headers);
+		DestinationInfo info = parseUserDestination(headers);
 		if (info == null) {
-			return Collections.emptySet();
+			return null;
 		}
 
-		Set<String> result = new HashSet<String>();
+		Set<String> targetDestinations = new HashSet<String>();
 		for (String sessionId : info.getSessionIds()) {
-			result.add(getTargetDestination(
-					headers.getDestination(), info.getDestination(), sessionId, info.getUser()));
+			targetDestinations.add(getTargetDestination(
+					headers.getDestination(), info.getDestinationWithoutPrefix(), sessionId, info.getUser()));
 		}
 
-		return result;
+		return new UserDestinationResult(headers.getDestination(),
+				targetDestinations, info.getSubscribeDestination(), info.getUser());
 	}
 
-	private UserDestinationInfo getUserDestinationInfo(SimpMessageHeaderAccessor headers) {
+	private DestinationInfo parseUserDestination(SimpMessageHeaderAccessor headers) {
 
 		String destination = headers.getDestination();
 
-		String targetUser;
-		String targetDestination;
-		Set<String> targetSessionIds;
+		String destinationWithoutPrefix;
+		String subscribeDestination;
+		String user;
+		Set<String> sessionIds;
 
 		Principal principal = headers.getUser();
 		SimpMessageType messageType = headers.getMessageType();
@@ -127,9 +140,10 @@ private UserDestinationInfo getUserDestinationInfo(SimpMessageHeaderAccessor hea
 				logger.error("Ignoring message, no session id available");
 				return null;
 			}
-			targetUser = principal.getName();
-			targetDestination = destination.substring(this.destinationPrefix.length()-1);
-			targetSessionIds = Collections.singleton(headers.getSessionId());
+			destinationWithoutPrefix = destination.substring(this.destinationPrefix.length()-1);
+			subscribeDestination = destination;
+			user = principal.getName();
+			sessionIds = Collections.singleton(headers.getSessionId());
 		}
 		else if (SimpMessageType.MESSAGE.equals(messageType)) {
 			if (!checkDestination(destination, this.destinationPrefix)) {
@@ -138,10 +152,11 @@ else if (SimpMessageType.MESSAGE.equals(messageType)) {
 			int startIndex = this.destinationPrefix.length();
 			int endIndex = destination.indexOf('/', startIndex);
 			Assert.isTrue(endIndex > 0, "Expected destination pattern \"/principal/{userId}/**\"");
-			targetUser = destination.substring(startIndex, endIndex);
-			targetUser = StringUtils.replace(targetUser, "%2F", "/");
-			targetDestination = destination.substring(endIndex);
-			targetSessionIds = this.userSessionRegistry.getSessionIds(targetUser);
+			destinationWithoutPrefix = destination.substring(endIndex);
+			subscribeDestination = this.destinationPrefix.substring(0, startIndex-1) + destinationWithoutPrefix;
+			user = destination.substring(startIndex, endIndex);
+			user = StringUtils.replace(user, "%2F", "/");
+			sessionIds = this.userSessionRegistry.getSessionIds(user);
 		}
 		else {
 			if (logger.isTraceEnabled()) {
@@ -150,7 +165,7 @@ else if (SimpMessageType.MESSAGE.equals(messageType)) {
 			return null;
 		}
 
-		return new UserDestinationInfo(targetUser, targetDestination, targetSessionIds);
+		return new DestinationInfo(destinationWithoutPrefix, subscribeDestination, user, sessionIds);
 	}
 
 	protected boolean checkDestination(String destination, String requiredPrefix) {
@@ -167,33 +182,53 @@ protected boolean checkDestination(String destination, String requiredPrefix) {
 		return true;
 	}
 
-	protected String getTargetDestination(String origDestination, String targetDestination,
-			String sessionId, String user) {
+	/**
+	 * Return the target destination to use. Provided as input are the original source
+	 * destination, as well as the same destination with the target prefix removed.
+	 *
+	 * @param sourceDestination the source destination from the input message
+	 * @param sourceDestinationWithoutPrefix the source destination with the target prefix removed
+	 * @param sessionId an active user session id
+	 * @param user the user
+	 * @return the target destination
+	 */
+	protected String getTargetDestination(String sourceDestination,
+			String sourceDestinationWithoutPrefix, String sessionId, String user) {
 
-		return targetDestination + "-user" + sessionId;
+		return sourceDestinationWithoutPrefix + "-user" + sessionId;
 	}
 
 
-	private static class UserDestinationInfo {
+	private static class DestinationInfo {
 
-		private final String user;
+		private final String destinationWithoutPrefix;
+
+		private final String subscribeDestination;
 
-		private final String destination;
+		private final String user;
 
 		private final Set<String> sessionIds;
 
-		private UserDestinationInfo(String user, String destination, Set<String> sessionIds) {
+
+		private DestinationInfo(String destinationWithoutPrefix, String subscribeDestination, String user,
+				Set<String> sessionIds) {
+
 			this.user = user;
-			this.destination = destination;
+			this.destinationWithoutPrefix = destinationWithoutPrefix;
+			this.subscribeDestination = subscribeDestination;
 			this.sessionIds = sessionIds;
 		}
 
-		public String getUser() {
-			return this.user;
+		public String getDestinationWithoutPrefix() {
+			return this.destinationWithoutPrefix;
 		}
 
-		public String getDestination() {
-			return this.destination;
+		public String getSubscribeDestination() {
+			return this.subscribeDestination;
+		}
+
+		public String getUser() {
+			return this.user;
 		}
 
 		public Set<String> getSessionIds() {

spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserDestinationMessageHandler.java

@@ -27,7 +27,11 @@
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.core.MessageSendingOperations;
+import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
+import org.springframework.messaging.simp.SimpMessageType;
 import org.springframework.messaging.simp.SimpMessagingTemplate;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.messaging.support.MessageHeaderAccessor;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 
@@ -43,6 +47,8 @@
  */
 public class UserDestinationMessageHandler implements MessageHandler, SmartLifecycle {
 
+	public static final String SUBSCRIBE_DESTINATION = "subscribeDestination";
+
 	private static final Log logger = LogFactory.getLog(UserDestinationMessageHandler.class);
 
 
@@ -140,11 +146,21 @@ public final void stop(Runnable callback) {
 	@Override
 	public void handleMessage(Message<?> message) throws MessagingException {
 
-		Set<String> destinations = this.userDestinationResolver.resolveDestination(message);
-		if (CollectionUtils.isEmpty(destinations)) {
+		UserDestinationResult result = this.userDestinationResolver.resolveDestination(message);
+		if (result == null) {
+			return;
+		}
+		Set<String> destinations = result.getTargetDestinations();
+		if (destinations.isEmpty()) {
 			return;
 		}
 
+		SimpMessageHeaderAccessor headerAccessor = SimpMessageHeaderAccessor.wrap(message);
+		if (SimpMessageType.MESSAGE.equals(headerAccessor.getMessageType())) {
+			headerAccessor.setHeader(SUBSCRIBE_DESTINATION, result.getSubscribeDestination());
+			message = MessageBuilder.withPayload(message.getPayload()).setHeaders(headerAccessor).build();
+		}
+
 		for (String targetDestination : destinations) {
 			if (logger.isDebugEnabled()) {
 				logger.debug("Sending message to resolved destination=" + targetDestination);

spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserDestinationResolver.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2014 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,38 +21,40 @@
 import java.util.Set;
 
 /**
- * A strategy for resolving unique, user destinations per session. User destinations
- * provide a user with the ability to subscribe to a queue unique to their session
- * as well others with the ability to send messages to those queues.
+ * A strategy for resolving a "user" destination and translating it to one or more
+ * actual destinations unique to the user's active session(s).
  * <p>
- * When a user attempts to subscribe to "/user/queue/position-updates", the
- * "/user" prefix is removed and a unique suffix added, resulting in something
- * like "/queue/position-updates-useri9oqdfzo" where the suffix is based on the
- * user's session and ensures it does not collide with any other users attempting
- * to subscribe to "/user/queue/position-updates".
+ * For messages sent to a user, the destination must contain the name of the target
+ * user, The name, extracted from the destination, is used to look up the active
+ * user session(s), and then translate the destination accordingly.
  * <p>
- * When a message is sent to a user with a destination such as
- * "/user/{username}/queue/position-updates", the "/user/{username}" prefix is
- * removed and the suffix added, resulting in something like
- * "/queue/position-updates-useri9oqdfzo".
+ * For SUBSCRIBE and UNSUBSCRIBE messages, the user is the user associated with
+ * the message. In other words the destination does not contain the user name.
+ * <p>
+ * See the documentation on implementations for specific examples.
  *
  * @author Rossen Stoyanchev
  * @since 4.0
  *
+ * @see org.springframework.messaging.simp.user.DefaultUserDestinationResolver
  * @see UserDestinationMessageHandler
  */
 public interface UserDestinationResolver {
 
 	/**
-	 * Resolve the destination of the message to a set of actual target destinations
-	 * to use. If the message is SUBSCRIBE/UNSUBSCRIBE, the returned set will contain
-	 * only target destination. If the message represents data being sent to a user,
-	 * the returned set may contain multiple target destinations, one for each active
-	 * session of the target user.
+	 * Resolve the destination of the message to a set of actual target destinations.
+	 * <p>
+	 * If the message is SUBSCRIBE/UNSUBSCRIBE, the returned set will contain a
+	 * single translated target destination.
+	 * <p>
+	 * If the message represents data being sent to a user, the returned set may
+	 * contain multiple target destinations, one for each active user session.
+	 *
+	 * @param message the message with a user destination to be resolved
 	 *
-	 * @param message the message to resolve
-	 * @return the resolved unique user destination(s) or an empty Set
+	 * @return the result of the resolution, or {@code null} if the resolution
+	 * 	fails (e.g. not a user destination, or no user info available, etc)
 	 */
-	Set<String> resolveDestination(Message<?> message);
+	UserDestinationResult resolveDestination(Message<?> message);
 
 }

spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserDestinationResult.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright 2002-2013 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.messaging.simp.user;
+
+import org.springframework.util.Assert;
+
+import java.util.Collections;
+import java.util.Set;
+
+/**
+ * A simple container for the result of parsing and translating a "user" destination
+ * in some source message into a set of actual target destinations by calling
+ * {@link org.springframework.messaging.simp.user.UserDestinationResolver}.
+ *
+ * @author Rossen Stoyanchev
+ * @since 4.0.2
+ */
+public class UserDestinationResult {
+
+	private final String sourceDestination;
+
+	private final Set<String> targetDestinations;
+
+	private final String subscribeDestination;
+
+	private final String user;
+
+
+	public UserDestinationResult(String sourceDestination,
+			Set<String> targetDestinations, String subscribeDestination, String user) {
+
+		Assert.notNull(sourceDestination, "'sourceDestination' must not be null");
+		Assert.notNull(targetDestinations, "'targetDestinations' must not be null");
+		Assert.notNull(subscribeDestination, "'subscribeDestination' must not be null");
+		Assert.notNull(user, "'user' must not be null");
+
+		this.sourceDestination = sourceDestination;
+		this.targetDestinations = targetDestinations;
+		this.subscribeDestination = subscribeDestination;
+		this.user = user;
+	}
+
+
+	/**
+	 * The "user" destination as found in the headers of the source message.
+	 *
+	 * @return a destination, never {@code null}
+	 */
+	public String getSourceDestination() {
+		return this.sourceDestination;
+	}
+
+	/**
+	 * The result of parsing the source destination and translating it into a set
+	 * of actual target destinations to use.
+	 *
+	 * @return a set of destination values, possibly an empty set
+	 */
+	public Set<String> getTargetDestinations() {
+		return this.targetDestinations;
+	}
+
+	/**
+	 * The canonical form of the user destination as would be required to subscribe.
+	 * This may be useful to ensure that messages received by clients contain the
+	 * original destination they used to subscribe.
+	 *
+	 * @return a destination, never {@code null}
+	 */
+	public String getSubscribeDestination() {
+		return this.subscribeDestination;
+	}
+
+	/**
+	 * The user associated with the user destination.
+	 *
+	 * @return the user name, never {@code null}
+	 */
+	public String getUser() {
+		return this.user;
+	}
+}