Fix issue with subscribe destination

The original fix for SPR-11423:
spring-projects@32e5f57

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.

Author: rstoyanchev

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

@@ -175,10 +175,8 @@ public void handleMessage(Message<?> message) throws MessagingException {
 		}
 		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 void handleMessage(Message<?> message) throws MessagingException {
 		}
 	}
 
+	private void initHeaders(SimpMessageHeaderAccessor headerAccessor) {
+		if (getHeaderInitializer() != null) {
+			getHeaderInitializer().initHeaders(headerAccessor);
+		}
+	}
+
 }

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

@@ -91,8 +91,9 @@ public void handleMessage() {
 		ArgumentCaptor<Message> 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));
 	}
 
 

spring-websocket/src/main/java/org/springframework/web/socket/messaging/StompSubProtocolHandler.java

@@ -298,10 +298,10 @@ else if (stompAccessor.getCommand() == null || StompCommand.SEND.equals(stompAcc
 				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)) {

spring-websocket/src/test/java/org/springframework/web/socket/messaging/StompSubProtocolHandlerTests.java

@@ -239,7 +239,7 @@ public void handleMessageToClientUserDestination() {
 		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<byte[]> message = MessageBuilder.createMessage(EMPTY_PAYLOAD, headers.getMessageHeaders());
 		this.protocolHandler.handleMessageToClient(this.session, message);