diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/user/DefaultUserDestinationResolver.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/user/DefaultUserDestinationResolver.java index 60a8e39f33..ec6a872c48 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/simp/user/DefaultUserDestinationResolver.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/user/DefaultUserDestinationResolver.java @@ -30,10 +30,21 @@ import java.util.HashSet; 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. *

- * 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". + *

+ * 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 class DefaultUserDestinationResolver implements UserDestinationResolver { } @Override - public Set 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 result = new HashSet(); + Set targetDestinations = new HashSet(); 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 targetSessionIds; + String destinationWithoutPrefix; + String subscribeDestination; + String user; + Set sessionIds; Principal principal = headers.getUser(); SimpMessageType messageType = headers.getMessageType(); @@ -127,9 +140,10 @@ public class DefaultUserDestinationResolver implements UserDestinationResolver { 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 @@ public class DefaultUserDestinationResolver implements UserDestinationResolver { 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 @@ public class DefaultUserDestinationResolver implements UserDestinationResolver { return null; } - return new UserDestinationInfo(targetUser, targetDestination, targetSessionIds); + return new DestinationInfo(destinationWithoutPrefix, subscribeDestination, user, sessionIds); } protected boolean checkDestination(String destination, String requiredPrefix) { @@ -167,35 +182,55 @@ public class DefaultUserDestinationResolver implements UserDestinationResolver { 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 destinationWithoutPrefix; + + private final String subscribeDestination; private final String user; - private final String destination; - private final Set sessionIds; - private UserDestinationInfo(String user, String destination, Set sessionIds) { + + private DestinationInfo(String destinationWithoutPrefix, String subscribeDestination, String user, + Set sessionIds) { + this.user = user; - this.destination = destination; + this.destinationWithoutPrefix = destinationWithoutPrefix; + this.subscribeDestination = subscribeDestination; this.sessionIds = sessionIds; } + public String getDestinationWithoutPrefix() { + return this.destinationWithoutPrefix; + } + + public String getSubscribeDestination() { + return this.subscribeDestination; + } + public String getUser() { return this.user; } - public String getDestination() { - return this.destination; - } - public Set getSessionIds() { return this.sessionIds; } diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserDestinationMessageHandler.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserDestinationMessageHandler.java index 011b83056a..5f0d1b1974 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserDestinationMessageHandler.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserDestinationMessageHandler.java @@ -27,7 +27,11 @@ import org.springframework.messaging.MessageHandler; 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 @@ import org.springframework.util.CollectionUtils; */ public class UserDestinationMessageHandler implements MessageHandler, SmartLifecycle { + public static final String SUBSCRIBE_DESTINATION = "subscribeDestination"; + private static final Log logger = LogFactory.getLog(UserDestinationMessageHandler.class); @@ -140,10 +146,20 @@ public class UserDestinationMessageHandler implements MessageHandler, SmartLifec @Override public void handleMessage(Message message) throws MessagingException { - Set destinations = this.userDestinationResolver.resolveDestination(message); - if (CollectionUtils.isEmpty(destinations)) { + UserDestinationResult result = this.userDestinationResolver.resolveDestination(message); + if (result == null) { return; } + Set 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()) { diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserDestinationResolver.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserDestinationResolver.java index f722c434c0..9204e9af92 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserDestinationResolver.java +++ b/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 org.springframework.messaging.Message; 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). *

- * 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. *

- * 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. + *

+ * 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. + *

+ * If the message is SUBSCRIBE/UNSUBSCRIBE, the returned set will contain a + * single translated target destination. + *

+ * 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 to resolve - * @return the resolved unique user destination(s) or an empty Set + * @param message the message with a user destination to be resolved + * + * @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 resolveDestination(Message message); + UserDestinationResult resolveDestination(Message message); } diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserDestinationResult.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserDestinationResult.java new file mode 100644 index 0000000000..d959be775d --- /dev/null +++ b/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 targetDestinations; + + private final String subscribeDestination; + + private final String user; + + + public UserDestinationResult(String sourceDestination, + Set 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 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; + } +} diff --git a/spring-messaging/src/test/java/org/springframework/messaging/simp/user/DefaultUserDestinationResolverTests.java b/spring-messaging/src/test/java/org/springframework/messaging/simp/user/DefaultUserDestinationResolverTests.java index 959cefccfe..6a8496f554 100644 --- a/spring-messaging/src/test/java/org/springframework/messaging/simp/user/DefaultUserDestinationResolverTests.java +++ b/spring-messaging/src/test/java/org/springframework/messaging/simp/user/DefaultUserDestinationResolverTests.java @@ -25,9 +25,8 @@ import org.springframework.messaging.simp.TestPrincipal; import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.StringUtils; -import java.util.Set; - import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; /** * Unit tests for {@link org.springframework.messaging.simp.user.DefaultUserDestinationResolver}. @@ -56,11 +55,15 @@ public class DefaultUserDestinationResolverTests { @Test public void handleSubscribe() { - Message message = createMessage(SimpMessageType.SUBSCRIBE, this.user, SESSION_ID, "/user/queue/foo"); - Set actual = this.resolver.resolveDestination(message); + String sourceDestination = "/user/queue/foo"; + Message message = createMessage(SimpMessageType.SUBSCRIBE, this.user, SESSION_ID, sourceDestination); + UserDestinationResult actual = this.resolver.resolveDestination(message); - assertEquals(1, actual.size()); - assertEquals("/queue/foo-user123", actual.iterator().next()); + assertEquals(sourceDestination, actual.getSourceDestination()); + assertEquals(1, actual.getTargetDestinations().size()); + assertEquals("/queue/foo-user123", actual.getTargetDestinations().iterator().next()); + assertEquals(sourceDestination, actual.getSubscribeDestination()); + assertEquals(this.user.getName(), actual.getUser()); } // SPR-11325 @@ -72,28 +75,32 @@ public class DefaultUserDestinationResolverTests { this.registry.registerSessionId("joe", "789"); Message message = createMessage(SimpMessageType.SUBSCRIBE, this.user, SESSION_ID, "/user/queue/foo"); - Set actual = this.resolver.resolveDestination(message); + UserDestinationResult actual = this.resolver.resolveDestination(message); - assertEquals(1, actual.size()); - assertEquals("/queue/foo-user123", actual.iterator().next()); + assertEquals(1, actual.getTargetDestinations().size()); + assertEquals("/queue/foo-user123", actual.getTargetDestinations().iterator().next()); } @Test public void handleUnsubscribe() { Message message = createMessage(SimpMessageType.UNSUBSCRIBE, this.user, SESSION_ID, "/user/queue/foo"); - Set actual = this.resolver.resolveDestination(message); + UserDestinationResult actual = this.resolver.resolveDestination(message); - assertEquals(1, actual.size()); - assertEquals("/queue/foo-user123", actual.iterator().next()); + assertEquals(1, actual.getTargetDestinations().size()); + assertEquals("/queue/foo-user123", actual.getTargetDestinations().iterator().next()); } @Test public void handleMessage() { - Message message = createMessage(SimpMessageType.MESSAGE, this.user, SESSION_ID, "/user/joe/queue/foo"); - Set actual = this.resolver.resolveDestination(message); + String sourceDestination = "/user/joe/queue/foo"; + Message message = createMessage(SimpMessageType.MESSAGE, this.user, SESSION_ID, sourceDestination); + UserDestinationResult actual = this.resolver.resolveDestination(message); - assertEquals(1, actual.size()); - assertEquals("/queue/foo-user123", actual.iterator().next()); + assertEquals(sourceDestination, actual.getSourceDestination()); + assertEquals(1, actual.getTargetDestinations().size()); + assertEquals("/queue/foo-user123", actual.getTargetDestinations().iterator().next()); + assertEquals("/user/queue/foo", actual.getSubscribeDestination()); + assertEquals(this.user.getName(), actual.getUser()); } @Test @@ -103,10 +110,10 @@ public class DefaultUserDestinationResolverTests { this.registry.registerSessionId(userName, "openid123"); String destination = "/user/" + StringUtils.replace(userName, "/", "%2F") + "/queue/foo"; Message message = createMessage(SimpMessageType.MESSAGE, this.user, SESSION_ID, destination); - Set actual = this.resolver.resolveDestination(message); + UserDestinationResult actual = this.resolver.resolveDestination(message); - assertEquals(1, actual.size()); - assertEquals("/queue/foo-useropenid123", actual.iterator().next()); + assertEquals(1, actual.getTargetDestinations().size()); + assertEquals("/queue/foo-useropenid123", actual.getTargetDestinations().iterator().next()); } @Test @@ -114,28 +121,28 @@ public class DefaultUserDestinationResolverTests { // no destination Message message = createMessage(SimpMessageType.MESSAGE, this.user, SESSION_ID, null); - Set actual = this.resolver.resolveDestination(message); - assertEquals(0, actual.size()); + UserDestinationResult actual = this.resolver.resolveDestination(message); + assertNull(actual); // not a user destination message = createMessage(SimpMessageType.MESSAGE, this.user, SESSION_ID, "/queue/foo"); actual = this.resolver.resolveDestination(message); - assertEquals(0, actual.size()); + assertNull(actual); // subscribe + no user message = createMessage(SimpMessageType.SUBSCRIBE, null, SESSION_ID, "/user/queue/foo"); actual = this.resolver.resolveDestination(message); - assertEquals(0, actual.size()); + assertNull(actual); // subscribe + not a user destination message = createMessage(SimpMessageType.SUBSCRIBE, this.user, SESSION_ID, "/queue/foo"); actual = this.resolver.resolveDestination(message); - assertEquals(0, actual.size()); + assertNull(actual); // no match on message type message = createMessage(SimpMessageType.CONNECT, this.user, SESSION_ID, "user/joe/queue/foo"); actual = this.resolver.resolveDestination(message); - assertEquals(0, actual.size()); + assertNull(actual); } diff --git a/spring-messaging/src/test/java/org/springframework/messaging/simp/user/UserDestinationMessageHandlerTests.java b/spring-messaging/src/test/java/org/springframework/messaging/simp/user/UserDestinationMessageHandlerTests.java index abd622ae92..717388af0d 100644 --- a/spring-messaging/src/test/java/org/springframework/messaging/simp/user/UserDestinationMessageHandlerTests.java +++ b/spring-messaging/src/test/java/org/springframework/messaging/simp/user/UserDestinationMessageHandlerTests.java @@ -95,6 +95,8 @@ public class UserDestinationMessageHandlerTests { assertEquals("/queue/foo-user123", captor.getValue().getHeaders().get(SimpMessageHeaderAccessor.DESTINATION_HEADER)); + assertEquals("/user/queue/foo", + captor.getValue().getHeaders().get(UserDestinationMessageHandler.SUBSCRIBE_DESTINATION)); } diff --git a/spring-websocket/src/main/java/org/springframework/web/socket/messaging/StompSubProtocolHandler.java b/spring-websocket/src/main/java/org/springframework/web/socket/messaging/StompSubProtocolHandler.java index 4e7f17f0e3..911e574a7e 100644 --- a/spring-websocket/src/main/java/org/springframework/web/socket/messaging/StompSubProtocolHandler.java +++ b/spring-websocket/src/main/java/org/springframework/web/socket/messaging/StompSubProtocolHandler.java @@ -36,6 +36,7 @@ import org.springframework.messaging.simp.stomp.StompDecoder; import org.springframework.messaging.simp.stomp.StompEncoder; import org.springframework.messaging.simp.stomp.StompHeaderAccessor; import org.springframework.messaging.simp.user.DestinationUserNameProvider; +import org.springframework.messaging.simp.user.UserDestinationMessageHandler; import org.springframework.messaging.simp.user.UserSessionRegistry; import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.Assert; @@ -180,9 +181,15 @@ public class StompSubProtocolHandler implements SubProtocolHandler { afterStompSessionConnected(headers, session); } - if (StompCommand.MESSAGE.equals(headers.getCommand()) && (headers.getSubscriptionId() == null)) { - logger.error("Ignoring message, no subscriptionId header: " + message); - return; + if (StompCommand.MESSAGE.equals(headers.getCommand())) { + if (headers.getSubscriptionId() == null) { + logger.error("Ignoring message, no subscriptionId header: " + message); + return; + } + String header = UserDestinationMessageHandler.SUBSCRIBE_DESTINATION; + if (message.getHeaders().containsKey(header)) { + headers.setDestination((String) message.getHeaders().get(header)); + } } if (!(message.getPayload() instanceof byte[])) { diff --git a/spring-websocket/src/test/java/org/springframework/web/socket/messaging/StompSubProtocolHandlerTests.java b/spring-websocket/src/test/java/org/springframework/web/socket/messaging/StompSubProtocolHandlerTests.java index e571b01faa..733a140cf6 100644 --- a/spring-websocket/src/test/java/org/springframework/web/socket/messaging/StompSubProtocolHandlerTests.java +++ b/spring-websocket/src/test/java/org/springframework/web/socket/messaging/StompSubProtocolHandlerTests.java @@ -35,6 +35,7 @@ import org.springframework.messaging.simp.stomp.StompDecoder; import org.springframework.messaging.simp.stomp.StompHeaderAccessor; import org.springframework.messaging.simp.user.DefaultUserSessionRegistry; import org.springframework.messaging.simp.user.DestinationUserNameProvider; +import org.springframework.messaging.simp.user.UserDestinationMessageHandler; import org.springframework.messaging.simp.user.UserSessionRegistry; import org.springframework.messaging.support.MessageBuilder; import org.springframework.web.socket.TextMessage; @@ -138,6 +139,22 @@ public class StompSubProtocolHandlerTests { assertEquals("joe", replyHeaders.getNativeHeader("user-name").get(0)); } + @Test + public void handleMessageToClientUserDestination() { + + StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.MESSAGE); + headers.setMessageId("mess0"); + headers.setSubscriptionId("sub0"); + headers.setDestination("/queue/foo-user123"); + headers.setHeader(UserDestinationMessageHandler.SUBSCRIBE_DESTINATION, "/user/queue/foo"); + Message message = MessageBuilder.withPayload(new byte[0]).setHeaders(headers).build(); + this.protocolHandler.handleMessageToClient(this.session, message); + + assertEquals(1, this.session.getSentMessages().size()); + WebSocketMessage textMessage = this.session.getSentMessages().get(0); + assertTrue(((String) textMessage.getPayload()).contains("destination:/user/queue/foo\n")); + } + @Test public void handleMessageFromClient() {