Update user destinations handling

Before this change subscribing to a user-specific destination in STOMP
required manually appending a unique queue suffix provided in a header
with the CONNECTED frame.

This change removes the need to do that. Instead STOMP clients can
subscribe to "/user/queue/error" and can then begin to receive messages
sent to "/user/{username}/queue/error" without colliding with any other
user doing the same.

Issue: SPR-11077

spring-messaging/src/test/java/org/springframework/messaging/simp/TestPrincipal.java

@@ -0,0 +1,57 @@
+/*
+ * 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;
+
+import java.security.Principal;
+
+
+/**
+ * An implementation of {@link Principal} for testing.
+ *
+ * @author Rossen Stoyanchev
+ */
+public class TestPrincipal implements Principal {
+
+	private String name;
+
+	public TestPrincipal(String name) {
+		this.name = name;
+	}
+
+	@Override
+	public String getName() {
+		return this.name;
+	}
+
+	@Override
+	public boolean equals(Object obj) {
+		if (obj == this) {
+			return true;
+		}
+		if (!(obj instanceof TestPrincipal)) {
+			return false;
+		}
+		TestPrincipal p = (TestPrincipal) obj;
+		return this.name.equals(p.name);
+	}
+
+	@Override
+	public int hashCode() {
+		return this.name.hashCode();
+	}
+
+}

spring-messaging/src/test/java/org/springframework/messaging/simp/config/ServletStompEndpointRegistryTests.java

@@ -24,8 +24,8 @@ import org.mockito.Mockito;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.handler.websocket.SubProtocolHandler;
 import org.springframework.messaging.handler.websocket.SubProtocolWebSocketHandler;
-import org.springframework.messaging.simp.handler.MutableUserQueueSuffixResolver;
-import org.springframework.messaging.simp.handler.SimpleUserQueueSuffixResolver;
+import org.springframework.messaging.simp.handler.DefaultUserSessionRegistry;
+import org.springframework.messaging.simp.handler.UserSessionRegistry;
 import org.springframework.messaging.simp.stomp.StompProtocolHandler;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.web.servlet.handler.SimpleUrlHandlerMapping;
@@ -44,16 +44,16 @@ public class ServletStompEndpointRegistryTests {
 
 	private SubProtocolWebSocketHandler webSocketHandler;
 
-	private MutableUserQueueSuffixResolver queueSuffixResolver;
+	private UserSessionRegistry userSessionRegistry;
 
 
 	@Before
 	public void setup() {
 		MessageChannel channel = Mockito.mock(MessageChannel.class);
 		this.webSocketHandler = new SubProtocolWebSocketHandler(channel);
-		this.queueSuffixResolver = new SimpleUserQueueSuffixResolver();
+		this.userSessionRegistry = new DefaultUserSessionRegistry();
 		TaskScheduler taskScheduler = Mockito.mock(TaskScheduler.class);
-		this.registry = new ServletStompEndpointRegistry(webSocketHandler, queueSuffixResolver, taskScheduler);
+		this.registry = new ServletStompEndpointRegistry(webSocketHandler, userSessionRegistry, taskScheduler);
 	}
 
 
@@ -69,7 +69,7 @@ public class ServletStompEndpointRegistryTests {
 		assertNotNull(protocolHandlers.get("v12.stomp"));
 
 		StompProtocolHandler stompHandler = (StompProtocolHandler) protocolHandlers.get("v10.stomp");
-		assertSame(this.queueSuffixResolver, stompHandler.getUserQueueSuffixResolver());
+		assertSame(this.userSessionRegistry, stompHandler.getUserSessionRegistry());
 	}
 
 	@Test

spring-messaging/src/test/java/org/springframework/messaging/simp/config/WebSocketMessageBrokerConfigurationSupportTests.java

@@ -35,9 +35,9 @@ import org.springframework.messaging.handler.websocket.SubProtocolWebSocketHandl
 import org.springframework.messaging.simp.SimpMessageType;
 import org.springframework.messaging.simp.annotation.SubscribeEvent;
 import org.springframework.messaging.simp.handler.SimpAnnotationMethodMessageHandler;
-import org.springframework.messaging.simp.handler.MutableUserQueueSuffixResolver;
 import org.springframework.messaging.simp.handler.SimpleBrokerMessageHandler;
 import org.springframework.messaging.simp.handler.UserDestinationMessageHandler;
+import org.springframework.messaging.simp.handler.UserSessionRegistry;
 import org.springframework.messaging.simp.stomp.StompBrokerRelayMessageHandler;
 import org.springframework.messaging.simp.stomp.StompCommand;
 import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
@@ -259,7 +259,7 @@ public class WebSocketMessageBrokerConfigurationSupportTests {
 		SubscribableChannel channel = this.cxtSimpleBroker.getBean("brokerChannel", SubscribableChannel.class);
 		UserDestinationMessageHandler messageHandler = this.cxtSimpleBroker.getBean(UserDestinationMessageHandler.class);
 
-		this.cxtSimpleBroker.getBean(MutableUserQueueSuffixResolver.class).addQueueSuffix("joe", "s1", "s1");
+		this.cxtSimpleBroker.getBean(UserSessionRegistry.class).registerSessionId("joe", "s1");
 
 		StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.SEND);
 		headers.setDestination("/user/joe/foo");
@@ -274,7 +274,7 @@ public class WebSocketMessageBrokerConfigurationSupportTests {
 		headers = StompHeaderAccessor.wrap(message);
 
 		assertEquals(SimpMessageType.MESSAGE, headers.getMessageType());
-		assertEquals("/foos1", headers.getDestination());
+		assertEquals("/foo-users1", headers.getDestination());
 	}
 
 	@Test

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

@@ -0,0 +1,120 @@
+/*
+ * 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.handler;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
+import org.springframework.messaging.simp.SimpMessageType;
+import org.springframework.messaging.simp.TestPrincipal;
+import org.springframework.messaging.support.MessageBuilder;
+
+import java.util.Set;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Unit tests for {@link DefaultUserDestinationResolver}.
+ */
+public class DefaultUserDestinationResolverTests {
+
+	private DefaultUserDestinationResolver resolver;
+
+	private UserSessionRegistry registry;
+
+
+	@Before
+	public void setup() {
+		this.registry = new DefaultUserSessionRegistry();
+		this.resolver = new DefaultUserDestinationResolver(this.registry);
+	}
+
+
+	@Test
+	public void handleSubscribe() {
+		Message<?> message = createMessage(SimpMessageType.SUBSCRIBE, "joe", "/user/queue/foo");
+		this.registry.registerSessionId("joe", "123");
+		Set<String> actual = this.resolver.resolveDestination(message);
+
+		assertEquals(1, actual.size());
+		assertEquals("/queue/foo-user123", actual.iterator().next());
+	}
+
+	@Test
+	public void handleUnsubscribe() {
+		Message<?> message = createMessage(SimpMessageType.UNSUBSCRIBE, "joe", "/user/queue/foo");
+		this.registry.registerSessionId("joe", "123");
+		Set<String> actual = this.resolver.resolveDestination(message);
+
+		assertEquals(1, actual.size());
+		assertEquals("/queue/foo-user123", actual.iterator().next());
+	}
+
+	@Test
+	public void handleMessage() {
+		Message<?> message = createMessage(SimpMessageType.MESSAGE, "joe", "/user/joe/queue/foo");
+		this.registry.registerSessionId("joe", "123");
+		Set<String> actual = this.resolver.resolveDestination(message);
+
+		assertEquals(1, actual.size());
+		assertEquals("/queue/foo-user123", actual.iterator().next());
+	}
+
+
+	@Test
+	public void ignoreMessage() {
+
+		// no destination
+		Message<?> message = createMessage(SimpMessageType.MESSAGE, "joe", null);
+		Set<String> actual = this.resolver.resolveDestination(message);
+		assertEquals(0, actual.size());
+
+		// not a user destination
+		message = createMessage(SimpMessageType.MESSAGE, "joe", "/queue/foo");
+		actual = this.resolver.resolveDestination(message);
+		assertEquals(0, actual.size());
+
+		// subscribe + no user
+		message = createMessage(SimpMessageType.SUBSCRIBE, null, "/user/queue/foo");
+		actual = this.resolver.resolveDestination(message);
+		assertEquals(0, actual.size());
+
+		// subscribe + not a user destination
+		message = createMessage(SimpMessageType.SUBSCRIBE, "joe", "/queue/foo");
+		actual = this.resolver.resolveDestination(message);
+		assertEquals(0, actual.size());
+
+		// no match on message type
+		message = createMessage(SimpMessageType.CONNECT, "joe", "user/joe/queue/foo");
+		actual = this.resolver.resolveDestination(message);
+		assertEquals(0, actual.size());
+	}
+
+
+	private Message<?> createMessage(SimpMessageType messageType, String user, String destination) {
+		SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.create(messageType);
+		if (destination != null) {
+			headers.setDestination(destination);
+		}
+		if (user != null) {
+			headers.setUser(new TestPrincipal(user));
+		}
+		return MessageBuilder.withPayload(new byte[0]).setHeaders(headers).build();
+	}
+
+}

spring-messaging/src/test/java/org/springframework/messaging/simp/handler/DefaultUserSessionRegistryTests.java

@@ -27,56 +27,57 @@ import static org.junit.Assert.*;
 
 
 /**
- * Test fixture for {@link SimpleUserQueueSuffixResolver}
+ * Test fixture for {@link DefaultUserSessionRegistry}
  *
  * @author Rossen Stoyanchev
  * @since 4.0
  */
-public class SimpleUserQueueSuffixResolverTests {
+public class DefaultUserSessionRegistryTests {
 
 	private static final String user = "joe";
+
 	private static final List<String> sessionIds = Arrays.asList("sess01", "sess02", "sess03");
 
 
 	@Test
 	public void addOneSessionId() {
 
-		SimpleUserQueueSuffixResolver resolver = new SimpleUserQueueSuffixResolver();
-		resolver.addQueueSuffix(user, sessionIds.get(0), sessionIds.get(0));
+		DefaultUserSessionRegistry resolver = new DefaultUserSessionRegistry();
+		resolver.registerSessionId(user, sessionIds.get(0));
 
-		assertEquals(Collections.singleton(sessionIds.get(0)), resolver.getUserQueueSuffixes(user));
-		assertSame(Collections.emptySet(), resolver.getUserQueueSuffixes("jane"));
+		assertEquals(Collections.singleton(sessionIds.get(0)), resolver.getSessionIds(user));
+		assertSame(Collections.emptySet(), resolver.getSessionIds("jane"));
 	}
 
 	@Test
 	public void addMultipleSessionIds() {
 
-		SimpleUserQueueSuffixResolver resolver = new SimpleUserQueueSuffixResolver();
+		DefaultUserSessionRegistry resolver = new DefaultUserSessionRegistry();
 		for (String sessionId : sessionIds) {
-			resolver.addQueueSuffix(user, sessionId, sessionId);
+			resolver.registerSessionId(user, sessionId);
 		}
 
-		assertEquals(new LinkedHashSet<>(sessionIds), resolver.getUserQueueSuffixes(user));
-		assertEquals(Collections.emptySet(), resolver.getUserQueueSuffixes("jane"));
+		assertEquals(new LinkedHashSet<>(sessionIds), resolver.getSessionIds(user));
+		assertEquals(Collections.emptySet(), resolver.getSessionIds("jane"));
 	}
 
 
 	@Test
 	public void removeSessionIds() {
 
-		SimpleUserQueueSuffixResolver resolver = new SimpleUserQueueSuffixResolver();
+		DefaultUserSessionRegistry resolver = new DefaultUserSessionRegistry();
 		for (String sessionId : sessionIds) {
-			resolver.addQueueSuffix(user, sessionId, sessionId);
+			resolver.registerSessionId(user, sessionId);
 		}
 
-		assertEquals(new LinkedHashSet<>(sessionIds), resolver.getUserQueueSuffixes(user));
+		assertEquals(new LinkedHashSet<>(sessionIds), resolver.getSessionIds(user));
 
-		resolver.removeQueueSuffix(user, sessionIds.get(1));
-		resolver.removeQueueSuffix(user, sessionIds.get(2));
-		assertEquals(Collections.singleton(sessionIds.get(0)), resolver.getUserQueueSuffixes(user));
+		resolver.unregisterSessionId(user, sessionIds.get(1));
+		resolver.unregisterSessionId(user, sessionIds.get(2));
+		assertEquals(Collections.singleton(sessionIds.get(0)), resolver.getSessionIds(user));
 
-		resolver.removeQueueSuffix(user, sessionIds.get(0));
-		assertSame(Collections.emptySet(), resolver.getUserQueueSuffixes(user));
+		resolver.unregisterSessionId(user, sessionIds.get(0));
+		assertSame(Collections.emptySet(), resolver.getSessionIds(user));
 	}
 
 }

spring-messaging/src/test/java/org/springframework/messaging/simp/handler/UserDestinationMessageHandlerTests.java

@@ -0,0 +1,126 @@
+/*
+ * 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.handler;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.core.MessageSendingOperations;
+import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
+import org.springframework.messaging.simp.SimpMessageType;
+import org.springframework.messaging.simp.TestPrincipal;
+import org.springframework.messaging.support.MessageBuilder;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Unit tests for {@link UserDestinationMessageHandler}.
+ */
+public class UserDestinationMessageHandlerTests {
+
+	private UserDestinationMessageHandler messageHandler;
+
+	private MessageSendingOperations<String> messagingTemplate;
+
+	private UserSessionRegistry registry;
+
+
+	@Before
+	public void setup() {
+		this.messagingTemplate = Mockito.mock(MessageSendingOperations.class);
+		this.registry = new DefaultUserSessionRegistry();
+		DefaultUserDestinationResolver resolver = new DefaultUserDestinationResolver(this.registry);
+		this.messageHandler = new UserDestinationMessageHandler(this.messagingTemplate, resolver);
+	}
+
+
+	@Test
+	public void handleSubscribe() {
+		this.registry.registerSessionId("joe", "123");
+		this.messageHandler.handleMessage(createMessage(SimpMessageType.SUBSCRIBE, "joe", "/user/queue/foo"));
+
+		ArgumentCaptor<String> captor1 = ArgumentCaptor.forClass(String.class);
+		ArgumentCaptor<Message> captor2 = ArgumentCaptor.forClass(Message.class);
+		Mockito.verify(this.messagingTemplate).send(captor1.capture(), captor2.capture());
+
+		assertEquals("/queue/foo-user123", captor1.getValue());
+	}
+
+	@Test
+	public void handleUnsubscribe() {
+		this.registry.registerSessionId("joe", "123");
+		this.messageHandler.handleMessage(createMessage(SimpMessageType.UNSUBSCRIBE, "joe", "/user/queue/foo"));
+
+		ArgumentCaptor<String> captor1 = ArgumentCaptor.forClass(String.class);
+		ArgumentCaptor<Message> captor2 = ArgumentCaptor.forClass(Message.class);
+		Mockito.verify(this.messagingTemplate).send(captor1.capture(), captor2.capture());
+
+		assertEquals("/queue/foo-user123", captor1.getValue());
+	}
+
+	@Test
+	public void handleMessage() {
+		this.registry.registerSessionId("joe", "123");
+		this.messageHandler.handleMessage(createMessage(SimpMessageType.MESSAGE, "joe", "/user/joe/queue/foo"));
+
+		ArgumentCaptor<String> captor1 = ArgumentCaptor.forClass(String.class);
+		ArgumentCaptor<Message> captor2 = ArgumentCaptor.forClass(Message.class);
+		Mockito.verify(this.messagingTemplate).send(captor1.capture(), captor2.capture());
+
+		assertEquals("/queue/foo-user123", captor1.getValue());
+	}
+
+
+	@Test
+	public void ignoreMessage() {
+
+		// no destination
+		this.messageHandler.handleMessage(createMessage(SimpMessageType.MESSAGE, "joe", null));
+		Mockito.verifyZeroInteractions(this.messagingTemplate);
+
+		// not a user destination
+		this.messageHandler.handleMessage(createMessage(SimpMessageType.MESSAGE, "joe", "/queue/foo"));
+		Mockito.verifyZeroInteractions(this.messagingTemplate);
+
+		// subscribe + no user
+		this.messageHandler.handleMessage(createMessage(SimpMessageType.SUBSCRIBE, null, "/user/queue/foo"));
+		Mockito.verifyZeroInteractions(this.messagingTemplate);
+
+		// subscribe + not a user destination
+		this.messageHandler.handleMessage(createMessage(SimpMessageType.SUBSCRIBE, "joe", "/queue/foo"));
+		Mockito.verifyZeroInteractions(this.messagingTemplate);
+
+		// no match on message type
+		this.messageHandler.handleMessage(createMessage(SimpMessageType.CONNECT, "joe", "user/joe/queue/foo"));
+		Mockito.verifyZeroInteractions(this.messagingTemplate);
+	}
+
+
+	private Message<?> createMessage(SimpMessageType messageType, String user, String destination) {
+		SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.create(messageType);
+		if (destination != null) {
+			headers.setDestination(destination);
+		}
+		if (user != null) {
+			headers.setUser(new TestPrincipal(user));
+		}
+		return MessageBuilder.withPayload(new byte[0]).setHeaders(headers).build();
+	}
+
+}

spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/StompProtocolHandlerTests.java

@@ -28,9 +28,9 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
 import org.springframework.messaging.simp.SimpMessageType;
+import org.springframework.messaging.simp.TestPrincipal;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.web.socket.TextMessage;
-import org.springframework.web.socket.support.TestPrincipal;
 import org.springframework.web.socket.support.TestWebSocketSession;
 
 import static org.junit.Assert.*;
@@ -90,7 +90,6 @@ public class StompProtocolHandlerTests {
 		assertEquals("1.1", replyHeaders.getVersion());
 		assertArrayEquals(new long[] {0, 0}, replyHeaders.getHeartbeat());
 		assertEquals("joe", replyHeaders.getNativeHeader("user-name").get(0));
-		assertEquals("s1", replyHeaders.getNativeHeader("queue-suffix").get(0));
 	}
 
 	@Test