Support DestinationVariable on RSocket handlers

Prior to this commit, the pattern destination variables were not set in
the message headers prior to calling the handler. In this case, the
`DestinationVariableMethodArgumentResolver` could not get the
destination variables from the message headers and resolve those as
handler arguments.

This commit mutates the message headers if the message destination
contains patterns.

Fixes gh-22776

spring-messaging/src/test/java/org/springframework/messaging/handler/annotation/support/reactive/MessageMappingMessageHandlerTests.java

@@ -39,10 +39,13 @@ import org.springframework.core.io.buffer.DataBufferFactory;
 import org.springframework.core.io.buffer.DefaultDataBufferFactory;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.handler.DestinationPatternsMessageCondition;
+import org.springframework.messaging.handler.annotation.DestinationVariable;
 import org.springframework.messaging.handler.annotation.MessageExceptionHandler;
 import org.springframework.messaging.handler.annotation.MessageMapping;
 import org.springframework.messaging.handler.invocation.reactive.TestEncoderMethodReturnValueHandler;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.messaging.support.MessageHeaderAccessor;
 import org.springframework.stereotype.Controller;
 
 import static java.nio.charset.StandardCharsets.*;
@@ -89,6 +92,13 @@ public class MessageMappingMessageHandlerTests {
 		verifyOutputContent(Collections.singletonList("abcdef::response"));
 	}
 
+	@Test
+	public void handleWithDestinationVariable() {
+		MessageMappingMessageHandler messsageHandler = initMesssageHandler();
+		messsageHandler.handleMessage(message("destination.test", "abcdef")).block(Duration.ofSeconds(5));
+		verifyOutputContent(Collections.singletonList("test::abcdef::response"));
+	}
+
 	@Test
 	public void handleException() {
 		MessageMappingMessageHandler messsageHandler = initMesssageHandler();
@@ -143,9 +153,11 @@ public class MessageMappingMessageHandlerTests {
 	}
 
 	private Message<?> message(String destination, String... content) {
-		return new GenericMessage<>(
-				Flux.fromIterable(Arrays.asList(content)).map(payload -> toDataBuffer(payload)),
-				Collections.singletonMap(DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER, destination));
+		Flux<DataBuffer> payload = Flux.fromIterable(Arrays.asList(content)).map(parts -> toDataBuffer(parts));
+		MessageHeaderAccessor headers = new MessageHeaderAccessor();
+		headers.setLeaveMutable(true);
+		headers.setHeader(DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER, destination);
+		return MessageBuilder.createMessage(payload, headers.getMessageHeaders());
 	}
 
 	private DataBuffer toDataBuffer(String payload) {
@@ -181,6 +193,11 @@ public class MessageMappingMessageHandlerTests {
 			return payload + "::response";
 		}
 
+		@MessageMapping("destination.{variable}")
+		String handleWithDestinationVariable(@DestinationVariable String variable, String payload) {
+			return variable + "::" + payload + "::response";
+		}
+
 		@MessageMapping("exception")
 		String handleAndThrow() {
 			throw new IllegalArgumentException("rejected");