Add singleSession attribute to @SendToUser

Added the ability to send a message only to one user session. Given a
user has two tabs open and the client sends a message to the server
from tab 1, it is now possible to send a reply message to only 1 tab
instead of the default mode of targetting all known user sessions.

Issue: SPR-11506

spring-messaging/src/main/java/org/springframework/messaging/simp/annotation/SendToUser.java

@@ -48,4 +48,11 @@
 	 */
 	String[] value() default {};
 
+	/**
+     * A flag indicating whether the message is to be sent to a particular user session.
+     *
+     */
+    boolean singleSession() default false;
+
+
 }

spring-messaging/src/main/java/org/springframework/messaging/simp/annotation/support/SendToMethodReturnValueHandler.java

@@ -148,7 +148,12 @@ public void handleReturnValue(Object returnValue, MethodParameter returnType, Me
 			String user = getUserName(message, headers);
 			String[] destinations = getTargetDestinations(sendToUser, message, this.defaultUserDestinationPrefix);
 			for (String destination : destinations) {
-				this.messagingTemplate.convertAndSendToUser(user, destination, returnValue, createHeaders(sessionId));
+				if (sendToUser.singleSession()) {
+					this.messagingTemplate.convertAndSendToUser(userName, destination, returnValue, createHeaders(sessionId));
+				}
+				else } 
+					this.messagingTemplate.convertAndSendToUser(userName, destination, returnValue);
+				}
 			}
 			return;
 		}

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

@@ -158,7 +158,12 @@ else if (SimpMessageType.MESSAGE.equals(messageType)) {
 			subscribeDestination = this.destinationPrefix.substring(0, startIndex-1) + destinationWithoutPrefix;
 			user = destination.substring(startIndex, endIndex);
 			user = StringUtils.replace(user, "%2F", "/");
-			sessionIds = this.userSessionRegistry.getSessionIds(user);
+			if (headers.getSessionId() == null){
+				sessionIds = this.userSessionRegistry.getSessionIds(user);				
+			} else {
+				sessionIds = Collections.singleton(headers.getSessionId());
+			}
+
 		}
 		else {
 			if (logger.isTraceEnabled()) {

spring-messaging/src/test/java/org/springframework/messaging/simp/annotation/support/SendToMethodReturnValueHandlerTests.java

@@ -74,7 +74,9 @@ public class SendToMethodReturnValueHandlerTests {
 	private MethodParameter sendToReturnType;
 	private MethodParameter sendToDefaultDestReturnType;
 	private MethodParameter sendToUserReturnType;
+	private MethodParameter sendToUserSingleSessionReturnType;
 	private MethodParameter sendToUserDefaultDestReturnType;
+	private MethodParameter sendToUserSingleSessionDefaultDestReturnType;
 
 
 	@Before
@@ -100,9 +102,15 @@ public void setup() throws Exception {
 
 		method = this.getClass().getDeclaredMethod("handleAndSendToUser");
 		this.sendToUserReturnType = new MethodParameter(method, -1);
+
+		method = this.getClass().getDeclaredMethod("handleAndSendToUserSingleSession");
+		this.sendToUserSingleSessionReturnType = new MethodParameter(method, -1);
 
 		method = this.getClass().getDeclaredMethod("handleAndSendToUserDefaultDestination");
 		this.sendToUserDefaultDestReturnType = new MethodParameter(method, -1);
+
+		method = this.getClass().getDeclaredMethod("handleAndSendToUserSingleSessionDefaultDestination");
+		this.sendToUserSingleSessionDefaultDestReturnType = new MethodParameter(method, -1);
 	}
 
 
@@ -211,6 +219,31 @@ public void sendToUser() throws Exception {
 
 		verify(this.messageChannel, times(2)).send(this.messageCaptor.capture());
 
+		Message<?> message = this.messageCaptor.getAllValues().get(0);
+		SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
+		assertNull(headers.getSessionId());
+		assertNull(headers.getSubscriptionId());
+		assertEquals("/user/" + user.getName() + "/dest1", headers.getDestination());
+
+		message = this.messageCaptor.getAllValues().get(1);
+		headers = SimpMessageHeaderAccessor.wrap(message);
+		assertNull(headers.getSessionId());
+		assertNull(headers.getSubscriptionId());
+		assertEquals("/user/" + user.getName() + "/dest2", headers.getDestination());
+	}
+
+	@Test
+	public void sendToUserSingleSession() throws Exception {
+
+		when(this.messageChannel.send(any(Message.class))).thenReturn(true);
+
+		String sessionId = "sess1";
+		TestUser user = new TestUser();
+		Message<?> inputMessage = createInputMessage(sessionId, "sub1", null, user);
+		this.handler.handleReturnValue(payloadContent, this.sendToUserSingleSessionReturnType, inputMessage);
+
+		verify(this.messageChannel, times(2)).send(this.messageCaptor.capture());
+
 		Message<?> message = this.messageCaptor.getAllValues().get(0);
 		SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
 		assertEquals(sessionId, headers.getSessionId());
@@ -257,6 +290,25 @@ public void sendToUserDefaultDestination() throws Exception {
 
 		verify(this.messageChannel, times(1)).send(this.messageCaptor.capture());
 
+		Message<?> message = this.messageCaptor.getAllValues().get(0);
+		SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
+		assertNull(headers.getSessionId());
+		assertNull(headers.getSubscriptionId());
+		assertEquals("/user/" + user.getName() + "/queue/dest", headers.getDestination());
+	}
+
+	@Test
+	public void sendToUserDefaultDestinationSingleSession() throws Exception {
+
+		when(this.messageChannel.send(any(Message.class))).thenReturn(true);
+
+		String sessionId = "sess1";
+		TestUser user = new TestUser();
+		Message<?> inputMessage = createInputMessage(sessionId, "sub1", "/dest", user);
+		this.handler.handleReturnValue(payloadContent, this.sendToUserSingleSessionDefaultDestReturnType, inputMessage);
+
+		verify(this.messageChannel, times(1)).send(this.messageCaptor.capture());
+
 		Message<?> message = this.messageCaptor.getAllValues().get(0);
 		SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
 		assertEquals(sessionId, headers.getSessionId());
@@ -342,10 +394,20 @@ public String handleAndSendTo() {
 	public String handleAndSendToUserDefaultDestination() {
 		return PAYLOAD;
 	}
+
+	@SendToUser(singleSession=true)
+	public String handleAndSendToUserSingleSessionDefaultDestination() {
+		return payloadContent;
+	}
 
 	@SendToUser({"/dest1", "/dest2"})
 	public String handleAndSendToUser() {
 		return PAYLOAD;
 	}
+
+	@SendToUser(value={"/dest1", "/dest2"}, singleSession=true)
+	public String handleAndSendToUserSingleSession() {
+		return payloadContent;
+	}
 
 }

spring-messaging/src/test/java/org/springframework/messaging/simp/user/DefaultUserDestinationResolverTests.java

@@ -109,7 +109,7 @@ public void handleMessageEncodedUserName() {
 		String userName = "http://joe.openid.example.org/";
 		this.registry.registerSessionId(userName, "openid123");
 		String destination = "/user/" + StringUtils.replace(userName, "/", "%2F") + "/queue/foo";
-		Message<?> message = createMessage(SimpMessageType.MESSAGE, this.user, SESSION_ID, destination);
+		Message<?> message = createMessage(SimpMessageType.MESSAGE, this.user, null, destination);
 		UserDestinationResult actual = this.resolver.resolveDestination(message);
 
 		assertEquals(1, actual.getTargetDestinations().size());