From 1200755125f97c2f1035586081645bb244b06304 Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Fri, 25 Apr 2014 12:26:04 -0400 Subject: [PATCH] Fix issue with subscribe destination The original fix for SPR-11423: https://github.com/spring-projects/spring-framework/commit/32e5f57e647022d9ea82c03670936bf31f8467de was insufficient when using an external broker since the original destination header has to be in the "native headers" map (i.e. with STOMP headers) in order to be included in messages broadcast by the broker. --- .../simp/user/UserDestinationMessageHandler.java | 12 ++++++++---- .../user/UserDestinationMessageHandlerTests.java | 5 +++-- .../socket/messaging/StompSubProtocolHandler.java | 6 +++--- .../messaging/StompSubProtocolHandlerTests.java | 2 +- 4 files changed, 15 insertions(+), 10 deletions(-) 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 fded86db5b..969924b6b2 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 @@ -175,10 +175,8 @@ public class UserDestinationMessageHandler implements MessageHandler, SmartLifec } if (SimpMessageType.MESSAGE.equals(SimpMessageHeaderAccessor.getMessageType(message.getHeaders()))) { SimpMessageHeaderAccessor headerAccessor = SimpMessageHeaderAccessor.wrap(message); - if (getHeaderInitializer() != null) { - getHeaderInitializer().initHeaders(headerAccessor); - } - headerAccessor.setHeader(SimpMessageHeaderAccessor.ORIGINAL_DESTINATION, result.getSubscribeDestination()); + initHeaders(headerAccessor); + headerAccessor.setNativeHeader(SimpMessageHeaderAccessor.ORIGINAL_DESTINATION, result.getSubscribeDestination()); message = MessageBuilder.createMessage(message.getPayload(), headerAccessor.getMessageHeaders()); } for (String destination : destinations) { @@ -189,4 +187,10 @@ public class UserDestinationMessageHandler implements MessageHandler, SmartLifec } } + private void initHeaders(SimpMessageHeaderAccessor headerAccessor) { + if (getHeaderInitializer() != null) { + getHeaderInitializer().initHeaders(headerAccessor); + } + } + } 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 120d4e2d2a..5712ece96b 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 @@ -91,8 +91,9 @@ public class UserDestinationMessageHandlerTests { ArgumentCaptor captor = ArgumentCaptor.forClass(Message.class); Mockito.verify(this.brokerChannel).send(captor.capture()); - assertEquals("/queue/foo-user123", SimpMessageHeaderAccessor.getDestination(captor.getValue().getHeaders())); - assertEquals("/user/queue/foo", captor.getValue().getHeaders().get(SimpMessageHeaderAccessor.ORIGINAL_DESTINATION)); + SimpMessageHeaderAccessor accessor = SimpMessageHeaderAccessor.wrap(captor.getValue()); + assertEquals("/queue/foo-user123", accessor.getDestination()); + assertEquals("/user/queue/foo", accessor.getFirstNativeHeader(SimpMessageHeaderAccessor.ORIGINAL_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 60a8bf2aaa..2d45fa4188 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 @@ -298,10 +298,10 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE logger.error("Ignoring message, no subscriptionId header: " + message); return; } - String header = SimpMessageHeaderAccessor.ORIGINAL_DESTINATION; - if (message.getHeaders().containsKey(header)) { + String origDestination = stompAccessor.getFirstNativeHeader(SimpMessageHeaderAccessor.ORIGINAL_DESTINATION); + if (origDestination != null) { stompAccessor = toMutableAccessor(stompAccessor, message); - stompAccessor.setDestination((String) message.getHeaders().get(header)); + stompAccessor.setDestination(origDestination); } } else if (StompCommand.CONNECTED.equals(command)) { 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 032aaae0a0..3f9ffb3d2f 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 @@ -239,7 +239,7 @@ public class StompSubProtocolHandlerTests { headers.setMessageId("mess0"); headers.setSubscriptionId("sub0"); headers.setDestination("/queue/foo-user123"); - headers.setHeader(StompHeaderAccessor.ORIGINAL_DESTINATION, "/user/queue/foo"); + headers.setNativeHeader(StompHeaderAccessor.ORIGINAL_DESTINATION, "/user/queue/foo"); Message message = MessageBuilder.createMessage(EMPTY_PAYLOAD, headers.getMessageHeaders()); this.protocolHandler.handleMessageToClient(this.session, message);