RSocket requests: Add per message metadata support

* Use convenient `Consumer` API to avoid external iteration for
`setupMetadata` in the `ClientRSocketConnector`
* Add `routeVars` and `metadata` support into the `RSocketOutboundGateway`
* Cover new options in the XML and Java DSL configurations

Author: artembilan

spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java

@@ -153,7 +153,7 @@ public void setSetupData(Object setupData) {
 	public void afterPropertiesSet() {
 		super.afterPropertiesSet();
 
-		RSocketRequester.Builder rsocketRequesterBuilder =
+		this.rsocketRequesterMono =
 				RSocketRequester.builder()
 						.dataMimeType(getDataMimeType())
 						.metadataMimeType(getMetadataMimeType())
@@ -162,10 +162,8 @@ public void afterPropertiesSet() {
 						.setupRoute(this.setupRoute, this.setupRouteVars)
 						.rsocketFactory(this.factoryConfigurer)
 						.rsocketFactory((rsocketFactory) ->
-								rsocketFactory.acceptor(this.rSocketMessageHandler.responder()));
-		this.setupMetadata.forEach(rsocketRequesterBuilder::setupMetadata);
-		this.rsocketRequesterMono =
-				rsocketRequesterBuilder
+								rsocketFactory.acceptor(this.rSocketMessageHandler.responder()))
+						.apply((builder) -> this.setupMetadata.forEach(builder::setupMetadata))
 						.connect(this.clientTransport)
 						.cache();
 	}

spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/RSocketOutboundGatewayParser.java

@@ -52,6 +52,7 @@ protected BeanDefinitionBuilder parseHandler(Element element, ParserContext pars
 		populateValueOrExpressionIfAny(builder, element, parserContext, "command");
 		populateValueOrExpressionIfAny(builder, element, parserContext, "publisher-element-type");
 		populateValueOrExpressionIfAny(builder, element, parserContext, "expected-response-type");
+		populateValueOrExpressionIfAny(builder, element, parserContext, "metadata");
 		return builder;
 	}
 

spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/RSocketOutboundGatewaySpec.java

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.rsocket.dsl;
 
+import java.util.Map;
 import java.util.function.Function;
 
 import org.springframework.expression.Expression;
@@ -25,6 +26,7 @@
 import org.springframework.integration.rsocket.ClientRSocketConnector;
 import org.springframework.integration.rsocket.outbound.RSocketOutboundGateway;
 import org.springframework.messaging.Message;
+import org.springframework.util.MimeType;
 
 /**
  * The {@link MessageHandlerSpec} implementation for the {@link RSocketOutboundGateway}.
@@ -35,6 +37,10 @@
  */
 public class RSocketOutboundGatewaySpec extends MessageHandlerSpec<RSocketOutboundGatewaySpec, RSocketOutboundGateway> {
 
+	RSocketOutboundGatewaySpec(String route, Object... routeVariables) {
+		this.target = new RSocketOutboundGateway(route, routeVariables);
+	}
+
 	RSocketOutboundGatewaySpec(Expression routeExpression) {
 		this.target = new RSocketOutboundGateway(routeExpression);
 	}
@@ -186,4 +192,40 @@ public RSocketOutboundGatewaySpec expectedResponseType(Expression expectedRespon
 		return this;
 	}
 
+	/**
+	 * Configure a {@link Function} to evaluate a metadata as a {@code Map<Object, MimeType>}
+	 * for RSocket request against request message.
+	 * @param metadataFunction the {@code Function} to use.
+	 * @param <P> the expected request message payload type.
+	 * @return the spec
+	 * @see RSocketOutboundGateway#setMetadataExpression(Expression)
+	 */
+	public <P> RSocketOutboundGatewaySpec metadata(Function<Message<P>, Map<Object, MimeType>> metadataFunction) {
+		return metadata(new FunctionExpression<>(metadataFunction));
+	}
+
+	/**
+	 Configure a SpEL expression to evaluate a metadata as a {@code Map<Object, MimeType>}
+	 * for RSocket request against request message.
+	 * @param metadataExpression the SpEL expression to use.
+	 * @return the spec
+	 * @see RSocketOutboundGateway#setMetadataExpression(Expression)
+	 */
+	public RSocketOutboundGatewaySpec metadata(String metadataExpression) {
+		return metadata(PARSER.parseExpression(metadataExpression));
+	}
+
+	/**
+	 * Configure a SpEL expression to evaluate a metadata as a {@code Map<Object, MimeType>}
+	 * for RSocket request against request message.
+	 * for RSocket request type at runtime against a request message.
+	 * @param metadataExpression the SpEL expression to use.
+	 * @return the spec
+	 * @see RSocketOutboundGateway#setMetadataExpression(Expression)
+	 */
+	public RSocketOutboundGatewaySpec metadata(Expression metadataExpression) {
+		this.target.setMetadataExpression(metadataExpression);
+		return this;
+	}
+
 }

spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/RSockets.java

@@ -19,7 +19,6 @@
 import java.util.function.Function;
 
 import org.springframework.expression.Expression;
-import org.springframework.expression.common.LiteralExpression;
 import org.springframework.integration.expression.FunctionExpression;
 import org.springframework.messaging.Message;
 
@@ -34,12 +33,13 @@ public final class RSockets {
 
 	/**
 	 * Create an {@link RSocketOutboundGatewaySpec} builder for request-reply gateway
-	 * based on provided {@code route}.
+	 * based on provided {@code route} and optional variables to expand route template.
 	 * @param route the {@code route} to send requests.
+	 * @param routeVariables the variables to expand route template.
 	 * @return the RSocketOutboundGatewaySpec instance
 	 */
-	public static RSocketOutboundGatewaySpec outboundGateway(String route) {
-		return outboundGateway(new LiteralExpression(route));
+	public static RSocketOutboundGatewaySpec outboundGateway(String route, Object... routeVariables) {
+		return new RSocketOutboundGatewaySpec(route, routeVariables);
 	}
 
 	/**

spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java

@@ -16,6 +16,8 @@
 
 package org.springframework.integration.rsocket.outbound;
 
+import java.util.Map;
+
 import org.reactivestreams.Publisher;
 
 import org.springframework.core.ParameterizedTypeReference;
@@ -31,6 +33,8 @@
 import org.springframework.messaging.rsocket.annotation.support.RSocketRequesterMethodArgumentResolver;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.MimeType;
 
 import reactor.core.publisher.Mono;
 
@@ -67,6 +71,8 @@ public class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler
 
 	private final Expression routeExpression;
 
+	private Object[] routeVars;
+
 	@Nullable
 	private ClientRSocketConnector clientRSocketConnector;
 
@@ -76,22 +82,29 @@ public class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler
 
 	private Expression expectedResponseTypeExpression = new ValueExpression<>(String.class);
 
+	private Expression metadataExpression;
+
 	private EvaluationContext evaluationContext;
 
 	@Nullable
 	private Mono<RSocketRequester> rsocketRequesterMono;
 
 	/**
-	 * Instantiate based on the provided RSocket endpoint {@code route}.
+	 * Instantiate based on the provided RSocket endpoint {@code route}
+	 * and optional variables to expand route template.
 	 * @param route the RSocket endpoint route to use.
+	 * @param routeVariables the variables to expand route template.
 	 */
-	public RSocketOutboundGateway(String route) {
+	public RSocketOutboundGateway(String route, Object... routeVariables) {
 		this(new ValueExpression<>(route));
+		this.routeVars = routeVariables;
 	}
 
 	/**
 	 * Instantiate based on the provided SpEL expression to evaluate an RSocket endpoint {@code route}
 	 * at runtime against a request message.
+	 * If route is a template and variables expansion is required, it is recommended to do that
+	 * in this expression evaluation, for example using some bean with an appropriate logic.
 	 * @param routeExpression the SpEL expression to use.
 	 */
 	public RSocketOutboundGateway(Expression routeExpression) {
@@ -173,6 +186,14 @@ public void setExpectedResponseTypeExpression(Expression expectedResponseTypeExp
 		this.expectedResponseTypeExpression = expectedResponseTypeExpression;
 	}
 
+	/**
+	 * Specify a SpEL expression to evaluate a metadata for RSocket request
+	 * as {@code Map<Object, MimeType>} against request message.
+	 * @param metadataExpression the expression for metadata.
+	 */
+	public void setMetadataExpression(Expression metadataExpression) {
+		this.metadataExpression = metadataExpression;
+	}
 
 	@Override
 	protected void doInit() {
@@ -205,13 +226,23 @@ protected Object handleRequestMessage(Message<?> requestMessage) {
 				.flatMap((responseSpec) -> performRequest(responseSpec, requestMessage));
 	}
 
+	@SuppressWarnings("unchecked")
 	private RSocketRequester.RequestSpec createRequestSpec(RSocketRequester rsocketRequester,
 			Message<?> requestMessage) {
 
 		String route = this.routeExpression.getValue(this.evaluationContext, requestMessage, String.class);
 		Assert.notNull(route, () -> "The 'routeExpression' [" + this.routeExpression + "] must not evaluate to null");
 
-		return rsocketRequester.route(route);
+		RSocketRequester.RequestSpec requestSpec = rsocketRequester.route(route, this.routeVars);
+		if (this.metadataExpression != null) {
+			Map<Object, MimeType> metadata =
+					this.metadataExpression.getValue(this.evaluationContext, requestMessage, Map.class);
+			if (!CollectionUtils.isEmpty(metadata)) {
+				requestSpec.metadata((spec) -> metadata.forEach(spec::metadata));
+			}
+		}
+
+		return requestSpec;
 	}
 
 	private RSocketRequester.ResponseSpec createResponseSpec(RSocketRequester.RequestSpec requestSpec,
@@ -228,15 +259,14 @@ private RSocketRequester.ResponseSpec createResponseSpec(RSocketRequester.Reques
 		}
 	}
 
-	@SuppressWarnings({ "rawtypes", "unchecked" })
 	private RSocketRequester.ResponseSpec responseSpecForPublisher(RSocketRequester.RequestSpec requestSpec,
 			Publisher<?> payload, Object publisherElementType) {
 
 		if (publisherElementType instanceof Class<?>) {
-			return requestSpec.data(payload, (Class) publisherElementType);
+			return requestSpec.data(payload, (Class<?>) publisherElementType);
 		}
 		else {
-			return requestSpec.data(payload, (ParameterizedTypeReference) publisherElementType);
+			return requestSpec.data(payload, (ParameterizedTypeReference<?>) publisherElementType);
 		}
 	}
 

spring-integration-rsocket/src/main/resources/org/springframework/integration/rsocket/config/spring-integration-rsocket-5.2.xsd

@@ -180,7 +180,7 @@
 						<xsd:annotation>
 							<xsd:documentation>
 								A SpEL expression to evaluate a 'Class' or 'ParameterizedTypeReference'
-								for for an RSocket response at runtime
+								for an RSocket response at runtime
 								against request message.
 								Mutually exclusive with 'expected-response-type'.
 							</xsd:documentation>
@@ -199,6 +199,14 @@
 							</xsd:appinfo>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="metadata-expression" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+								A SpEL expression to evaluate a 'Map' representing a metadata
+								for an RSocket request at runtime against request message.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>

spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/config/RSocketOutboundGatewayParserTests-context.xml

@@ -21,6 +21,7 @@
 								  route-expression="'testRoute'"
 								  request-channel="requestChannel"
 								  publisher-element-type="byte[]"
-								  expected-response-type="java.util.Date"/>
+								  expected-response-type="java.util.Date"
+								  metadata-expression="{'metadata': new org.springframework.util.MimeType('*')}"/>
 
 </beans>

spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/config/RSocketOutboundGatewayParserTests.java

@@ -18,14 +18,18 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.util.Collections;
+
 import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.expression.Expression;
 import org.springframework.integration.rsocket.ClientRSocketConnector;
 import org.springframework.integration.rsocket.outbound.RSocketOutboundGateway;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+import org.springframework.util.MimeType;
 
 /**
  * @author Artem Bilan
@@ -34,7 +38,7 @@
  */
 @SpringJUnitConfig
 @DirtiesContext
-public class RSocketOutboundGatewayParserTests {
+class RSocketOutboundGatewayParserTests {
 
 	@Autowired
 	private ClientRSocketConnector clientRSocketConnector;
@@ -54,6 +58,10 @@ void testOutboundGatewayParser() {
 				.isEqualTo("byte[]");
 		assertThat(TestUtils.getPropertyValue(this.outboundGateway, "expectedResponseTypeExpression.literalValue"))
 				.isEqualTo("java.util.Date");
+		Expression metadataExpression =
+				TestUtils.getPropertyValue(this.outboundGateway, "metadataExpression", Expression.class);
+		assertThat(metadataExpression.getValue())
+				.isEqualTo(Collections.singletonMap("metadata", new MimeType("*")));
 	}
 
 }