From a95c76e4d7ae9b26c5fed2b79c77307318c65bb3 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 3 Sep 2019 15:56:39 -0400 Subject: [PATCH] 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 --- .../rsocket/ClientRSocketConnector.java | 8 ++-- .../config/RSocketOutboundGatewayParser.java | 1 + .../dsl/RSocketOutboundGatewaySpec.java | 42 +++++++++++++++++++ .../integration/rsocket/dsl/RSockets.java | 8 ++-- .../outbound/RSocketOutboundGateway.java | 42 ++++++++++++++++--- .../config/spring-integration-rsocket-5.2.xsd | 10 ++++- ...cketOutboundGatewayParserTests-context.xml | 3 +- .../RSocketOutboundGatewayParserTests.java | 10 ++++- 8 files changed, 106 insertions(+), 18 deletions(-) diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java index 6aa338231b..188dbd4f0c 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java @@ -153,7 +153,7 @@ public class ClientRSocketConnector extends AbstractRSocketConnector { public void afterPropertiesSet() { super.afterPropertiesSet(); - RSocketRequester.Builder rsocketRequesterBuilder = + this.rsocketRequesterMono = RSocketRequester.builder() .dataMimeType(getDataMimeType()) .metadataMimeType(getMetadataMimeType()) @@ -162,10 +162,8 @@ public class ClientRSocketConnector extends AbstractRSocketConnector { .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(); } diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/RSocketOutboundGatewayParser.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/RSocketOutboundGatewayParser.java index 60cdb9d53c..96159c04c3 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/RSocketOutboundGatewayParser.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/RSocketOutboundGatewayParser.java @@ -52,6 +52,7 @@ public class RSocketOutboundGatewayParser extends AbstractConsumerEndpointParser 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; } diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/RSocketOutboundGatewaySpec.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/RSocketOutboundGatewaySpec.java index 68dfc3326f..260dba72d3 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/RSocketOutboundGatewaySpec.java +++ b/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.expression.ValueExpression; 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 @@ import org.springframework.messaging.Message; */ public class RSocketOutboundGatewaySpec extends MessageHandlerSpec { + RSocketOutboundGatewaySpec(String route, Object... routeVariables) { + this.target = new RSocketOutboundGateway(route, routeVariables); + } + RSocketOutboundGatewaySpec(Expression routeExpression) { this.target = new RSocketOutboundGateway(routeExpression); } @@ -186,4 +192,40 @@ public class RSocketOutboundGatewaySpec extends MessageHandlerSpec} + * for RSocket request against request message. + * @param metadataFunction the {@code Function} to use. + * @param

the expected request message payload type. + * @return the spec + * @see RSocketOutboundGateway#setMetadataExpression(Expression) + */ + public

RSocketOutboundGatewaySpec metadata(Function, Map> metadataFunction) { + return metadata(new FunctionExpression<>(metadataFunction)); + } + + /** + Configure a SpEL expression to evaluate a metadata as a {@code Map} + * 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} + * 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; + } + } diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/RSockets.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/RSockets.java index 6a46c78724..75ef513547 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/RSockets.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/RSockets.java @@ -19,7 +19,6 @@ package org.springframework.integration.rsocket.dsl; 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); } /** diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java index 35d4bbe026..143587cd3f 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java +++ b/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.RSocketRequester; 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 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 class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler this.expectedResponseTypeExpression = expectedResponseTypeExpression; } + /** + * Specify a SpEL expression to evaluate a metadata for RSocket request + * as {@code Map} 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 @@ public class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler .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 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 @@ public class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler } } - @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); } } diff --git a/spring-integration-rsocket/src/main/resources/org/springframework/integration/rsocket/config/spring-integration-rsocket-5.2.xsd b/spring-integration-rsocket/src/main/resources/org/springframework/integration/rsocket/config/spring-integration-rsocket-5.2.xsd index 8f16688f8c..efdbefbc6f 100644 --- a/spring-integration-rsocket/src/main/resources/org/springframework/integration/rsocket/config/spring-integration-rsocket-5.2.xsd +++ b/spring-integration-rsocket/src/main/resources/org/springframework/integration/rsocket/config/spring-integration-rsocket-5.2.xsd @@ -180,7 +180,7 @@ 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'. @@ -199,6 +199,14 @@ + + + + A SpEL expression to evaluate a 'Map' representing a metadata + for an RSocket request at runtime against request message. + + + diff --git a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/config/RSocketOutboundGatewayParserTests-context.xml b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/config/RSocketOutboundGatewayParserTests-context.xml index 2a21d6e7a5..5aa2f699d6 100644 --- a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/config/RSocketOutboundGatewayParserTests-context.xml +++ b/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('*')}"/> diff --git a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/config/RSocketOutboundGatewayParserTests.java b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/config/RSocketOutboundGatewayParserTests.java index ce6ff83107..a89958233b 100644 --- a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/config/RSocketOutboundGatewayParserTests.java +++ b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/config/RSocketOutboundGatewayParserTests.java @@ -18,14 +18,18 @@ package org.springframework.integration.rsocket.config; 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 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; */ @SpringJUnitConfig @DirtiesContext -public class RSocketOutboundGatewayParserTests { +class RSocketOutboundGatewayParserTests { @Autowired private ClientRSocketConnector clientRSocketConnector; @@ -54,6 +58,10 @@ public class RSocketOutboundGatewayParserTests { .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("*"))); } }