Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

RSocket requests: Add per message metadata support

Browse Source 
* 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
This commit is contained in:
Artem Bilan
2019-09-03 15:56:39 -04:00
parent 444c1f9913
commit a95c76e4d7
8 changed files with 106 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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();
}

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

@@ -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;
}

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

@@ -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, 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 class RSocketOutboundGatewaySpec extends MessageHandlerSpec<RSocketOutbou
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;
}
}

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

@@ -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);
}
/**

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

@@ -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<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 class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler
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 @@ 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<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 @@ 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);
}
}

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

@@ -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>

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

@@ -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>

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

@@ -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("*")));
}
}