From 3ddc004a8be6eaddd860abdd36cf630437df4de5 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 14 Jun 2019 16:15:12 -0400 Subject: [PATCH] Fix RSocket adapters for routing metadata support --- .../rsocket/AbstractRSocketConnector.java | 17 +++++ .../rsocket/ClientRSocketConnector.java | 7 +- .../rsocket/IntegrationRSocket.java | 70 ++++++++++++++----- .../rsocket/IntegrationRSocketAcceptor.java | 48 ++++++++++--- .../rsocket/ServerRSocketConnector.java | 12 +--- ...RSocketInboundGatewayIntegrationTests.java | 3 + ...SocketOutboundGatewayIntegrationTests.java | 1 + 7 files changed, 119 insertions(+), 39 deletions(-) diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/AbstractRSocketConnector.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/AbstractRSocketConnector.java index 068d48bfae..a2eb620934 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/AbstractRSocketConnector.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/AbstractRSocketConnector.java @@ -52,6 +52,8 @@ public abstract class AbstractRSocketConnector private MimeType dataMimeType = MimeTypeUtils.TEXT_PLAIN; + private MimeType metadataMimeType = IntegrationRSocket.COMPOSITE_METADATA; + private RSocketStrategies rsocketStrategies = RSocketStrategies.builder() .decoder(StringDecoder.allMimeTypes()) @@ -80,6 +82,20 @@ public abstract class AbstractRSocketConnector return this.dataMimeType; } + /** + * Configure a {@link MimeType} for metadata exchanging. + * Default to {@code "message/x.rsocket.composite-metadata.v0"}. + * @param metadataMimeType the {@link MimeType} to use. + */ + public void setMetadataMimeType(MimeType metadataMimeType) { + Assert.notNull(metadataMimeType, "'metadataMimeType' must not be null"); + this.metadataMimeType = metadataMimeType; + } + + protected MimeType getMetadataMimeType() { + return this.metadataMimeType; + } + /** * Configure a {@link RSocketStrategies} for data encoding/decoding. * @param rsocketStrategies the {@link RSocketStrategies} to use. @@ -121,6 +137,7 @@ public abstract class AbstractRSocketConnector @Override public void afterPropertiesSet() { this.rsocketAcceptor.setDefaultDataMimeType(this.dataMimeType); + this.rsocketAcceptor.setDefaultMetadataMimeType(this.metadataMimeType); this.rsocketAcceptor.setRSocketStrategies(this.rsocketStrategies); this.rsocketAcceptor.afterPropertiesSet(); } 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 9b75b80aba..4886c9c765 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 @@ -122,7 +122,8 @@ public class ClientRSocketConnector extends AbstractRSocketConnector { super.afterPropertiesSet(); RSocketFactory.ClientRSocketFactory clientFactory = RSocketFactory.connect() - .dataMimeType(getDataMimeType().toString()); + .dataMimeType(getDataMimeType().toString()) + .metadataMimeType(getMetadataMimeType().toString()); this.factoryConfigurer.accept(clientFactory); clientFactory.acceptor(this.rsocketAcceptor); Payload connectPayload = EmptyPayload.INSTANCE; @@ -161,7 +162,9 @@ public class ClientRSocketConnector extends AbstractRSocketConnector { public Mono getRSocketRequester() { return this.rsocketMono - .map((rsocket) -> RSocketRequester.wrap(rsocket, getDataMimeType(), getRSocketStrategies())) + .map((rsocket) -> + RSocketRequester + .wrap(rsocket, getDataMimeType(), getMetadataMimeType(), getRSocketStrategies())) .cache(); } diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java index ede9e1a2c1..550f086b01 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java @@ -16,6 +16,9 @@ package org.springframework.integration.rsocket; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.List; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Function; @@ -38,10 +41,12 @@ import org.springframework.messaging.support.MessageBuilder; import org.springframework.messaging.support.MessageHeaderAccessor; import org.springframework.util.Assert; import org.springframework.util.MimeType; +import org.springframework.util.RouteMatcher; import io.netty.buffer.ByteBuf; import io.rsocket.AbstractRSocket; import io.rsocket.Payload; +import io.rsocket.metadata.CompositeMetadata; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.MonoProcessor; @@ -63,28 +68,44 @@ import reactor.core.publisher.MonoProcessor; */ class IntegrationRSocket extends AbstractRSocket { + static final MimeType COMPOSITE_METADATA = new MimeType("message", "x.rsocket.composite-metadata.v0"); + + static final MimeType ROUTING = new MimeType("message", "x.rsocket.routing.v0"); + + static final List METADATA_MIME_TYPES = Arrays.asList(COMPOSITE_METADATA, ROUTING); + + private final Function, Mono> handler; + private final RouteMatcher routeMatcher; + private final RSocketRequester requester; private final DataBufferFactory bufferFactory; - @Nullable - private MimeType dataMimeType; + private final MimeType dataMimeType; - IntegrationRSocket(Function, Mono> handler, RSocketRequester requester, - @Nullable MimeType defaultDataMimeType, DataBufferFactory bufferFactory) { + private final MimeType metadataMimeType; + + IntegrationRSocket(Function, Mono> handler, RouteMatcher routeMatcher, + RSocketRequester requester, MimeType dataMimeType, MimeType metadataMimeType, + DataBufferFactory bufferFactory) { Assert.notNull(handler, "'handler' is required"); + Assert.notNull(routeMatcher, "'routeMatcher' is required"); Assert.notNull(requester, "'requester' is required"); - this.handler = handler; - this.requester = requester; - this.dataMimeType = defaultDataMimeType; - this.bufferFactory = bufferFactory; - } + Assert.notNull(dataMimeType, "'dataMimeType' is required"); + Assert.notNull(metadataMimeType, "'metadataMimeType' is required"); - public void setDataMimeType(MimeType dataMimeType) { + Assert.isTrue(METADATA_MIME_TYPES.contains(metadataMimeType), + () -> "Unexpected metadatata mime type: '" + metadataMimeType + "'"); + + this.handler = handler; + this.routeMatcher = routeMatcher; + this.requester = requester; this.dataMimeType = dataMimeType; + this.metadataMimeType = metadataMimeType; + this.bufferFactory = bufferFactory; } public RSocketRequester getRequester() { @@ -165,6 +186,24 @@ class IntegrationRSocket extends AbstractRSocket { : Mono.error(new IllegalStateException("Something went wrong: reply Mono not set")))); } + String getDestination(Payload payload) { + if (this.metadataMimeType.equals(COMPOSITE_METADATA)) { + CompositeMetadata metadata = new CompositeMetadata(payload.metadata(), false); + for (CompositeMetadata.Entry entry : metadata) { + String mimeType = entry.getMimeType(); + if (ROUTING.toString().equals(mimeType)) { + return entry.getContent().toString(StandardCharsets.UTF_8); + } + } + return ""; + } + else if (this.metadataMimeType.equals(ROUTING)) { + return payload.getMetadataUtf8(); + } + // Should not happen (given constructor assertions) + throw new IllegalArgumentException("Unexpected metadata MimeType"); + } + private DataBuffer retainDataAndReleasePayload(Payload payload) { return payloadToDataBuffer(payload, this.bufferFactory); } @@ -172,10 +211,9 @@ class IntegrationRSocket extends AbstractRSocket { private MessageHeaders createHeaders(String destination, @Nullable MonoProcessor replyMono) { MessageHeaderAccessor headers = new MessageHeaderAccessor(); headers.setLeaveMutable(true); - headers.setHeader(DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER, destination); - if (this.dataMimeType != null) { - headers.setContentType(this.dataMimeType); - } + RouteMatcher.Route route = this.routeMatcher.parseRoute(destination); + headers.setHeader(DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER, route); + headers.setContentType(this.dataMimeType); headers.setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, this.requester); if (replyMono != null) { headers.setHeader(RSocketPayloadReturnValueHandler.RESPONSE_HEADER, replyMono); @@ -184,10 +222,6 @@ class IntegrationRSocket extends AbstractRSocket { return headers.getMessageHeaders(); } - static String getDestination(Payload payload) { - return payload.getMetadataUtf8(); - } - static DataBuffer payloadToDataBuffer(Payload payload, DataBufferFactory bufferFactory) { payload.retain(); try { diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketAcceptor.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketAcceptor.java index 2a37bea3c1..e11e0ae66b 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketAcceptor.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketAcceptor.java @@ -19,7 +19,7 @@ package org.springframework.integration.rsocket; import java.lang.reflect.Method; import java.util.Collections; import java.util.List; -import java.util.function.Function; +import java.util.function.BiFunction; import java.util.function.Predicate; import org.springframework.context.ApplicationContext; @@ -34,9 +34,13 @@ import org.springframework.messaging.handler.invocation.reactive.SyncHandlerMeth import org.springframework.messaging.rsocket.RSocketMessageHandler; import org.springframework.messaging.rsocket.RSocketRequester; import org.springframework.messaging.rsocket.RSocketStrategies; +import org.springframework.util.Assert; import org.springframework.util.MimeType; +import org.springframework.util.MimeTypeUtils; import org.springframework.util.ReflectionUtils; +import org.springframework.util.StringUtils; +import io.rsocket.ConnectionSetupPayload; import io.rsocket.RSocket; /** @@ -54,7 +58,8 @@ import io.rsocket.RSocket; * * @see org.springframework.messaging.rsocket.MessageHandlerAcceptor */ -class IntegrationRSocketAcceptor extends RSocketMessageHandler implements Function { +class IntegrationRSocketAcceptor extends RSocketMessageHandler + implements BiFunction { private static final Method HANDLE_MESSAGE_METHOD = ReflectionUtils.findMethod(ReactiveMessageHandler.class, "handleMessage", Message.class); @@ -62,6 +67,8 @@ class IntegrationRSocketAcceptor extends RSocketMessageHandler implements Functi @Nullable private MimeType defaultDataMimeType; + private MimeType defaultMetadataMimeType = IntegrationRSocket.COMPOSITE_METADATA; + /** * Configure the default content type to use for data payloads. *

By default this is not set. However a server acceptor will use the @@ -73,6 +80,17 @@ class IntegrationRSocketAcceptor extends RSocketMessageHandler implements Functi this.defaultDataMimeType = defaultDataMimeType; } + /** + * Configure the default {@code MimeType} for payload data if the + * {@code SETUP} frame did not specify one. + *

By default this is set to {@code "message/x.rsocket.composite-metadata.v0"} + * @param mimeType the MimeType to use + */ + public void setDefaultMetadataMimeType(MimeType mimeType) { + Assert.notNull(mimeType, "'metadataMimeType' is required"); + this.defaultMetadataMimeType = mimeType; + } + public boolean detectEndpoints() { ApplicationContext applicationContext = getApplicationContext(); if (applicationContext != null && getHandlerMethods().isEmpty()) { @@ -91,7 +109,7 @@ class IntegrationRSocketAcceptor extends RSocketMessageHandler implements Functi public void addEndpoint(IntegrationRSocketEndpoint endpoint) { registerHandlerMethod(endpoint, HANDLE_MESSAGE_METHOD, new CompositeMessageCondition( - new DestinationPatternsMessageCondition(endpoint.getPath(), getPathMatcher()))); + new DestinationPatternsMessageCondition(endpoint.getPath(), getRouteMatcher()))); } @Override @@ -105,16 +123,26 @@ class IntegrationRSocketAcceptor extends RSocketMessageHandler implements Functi } @Override - public RSocket apply(RSocket sendingRSocket) { - return createRSocket(sendingRSocket); + public RSocket apply(ConnectionSetupPayload setupPayload, RSocket sendingRSocket) { + return createRSocket(setupPayload, sendingRSocket); } - protected IntegrationRSocket createRSocket(RSocket rsocket) { + protected IntegrationRSocket createRSocket(ConnectionSetupPayload setupPayload, RSocket rsocket) { RSocketStrategies rsocketStrategies = getRSocketStrategies(); - return new IntegrationRSocket(this::handleMessage, - RSocketRequester.wrap(rsocket, this.defaultDataMimeType, rsocketStrategies), - this.defaultDataMimeType, - rsocketStrategies.dataBufferFactory()); + MimeType dataMimeType = + StringUtils.hasText(setupPayload.dataMimeType()) + ? MimeTypeUtils.parseMimeType(setupPayload.dataMimeType()) + : this.defaultDataMimeType; + Assert.notNull(dataMimeType, "No `dataMimeType` in the ConnectionSetupPayload and no default value"); + + MimeType metadataMimeType = + StringUtils.hasText(setupPayload.dataMimeType()) + ? MimeTypeUtils.parseMimeType(setupPayload.metadataMimeType()) + : this.defaultMetadataMimeType; + Assert.notNull(dataMimeType, "No `metadataMimeType` in the ConnectionSetupPayload and no default value"); + return new IntegrationRSocket(this::handleMessage, getRouteMatcher(), + RSocketRequester.wrap(rsocket, dataMimeType, metadataMimeType, rsocketStrategies), + dataMimeType, metadataMimeType, rsocketStrategies.dataBufferFactory()); } private static final class MessageHandlerMethodArgumentResolver implements SyncHandlerMethodArgumentResolver { diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketConnector.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketConnector.java index 93a507f804..558a7a9baa 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketConnector.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketConnector.java @@ -32,8 +32,6 @@ import org.springframework.core.io.buffer.DataBufferUtils; import org.springframework.lang.Nullable; import org.springframework.messaging.rsocket.RSocketRequester; import org.springframework.util.Assert; -import org.springframework.util.MimeTypeUtils; -import org.springframework.util.StringUtils; import io.rsocket.Closeable; import io.rsocket.ConnectionSetupPayload; @@ -162,7 +160,7 @@ public class ServerRSocketConnector extends AbstractRSocketConnector private static class ServerRSocketAcceptor extends IntegrationRSocketAcceptor implements SocketAcceptor { - private static final Log LOGGER = LogFactory.getLog(IntegrationRSocket.class); + private static final Log LOGGER = LogFactory.getLog(ServerRSocketAcceptor.class); private final Map clientRSocketRequesters = new HashMap<>(); @@ -172,16 +170,12 @@ public class ServerRSocketConnector extends AbstractRSocketConnector @Override public Mono accept(ConnectionSetupPayload setupPayload, RSocket sendingRSocket) { - String destination = IntegrationRSocket.getDestination(setupPayload); DataBuffer dataBuffer = IntegrationRSocket.payloadToDataBuffer(setupPayload, getRSocketStrategies().dataBufferFactory()); int refCount = IntegrationRSocket.refCount(dataBuffer); - return Mono.just(sendingRSocket) - .map(this::createRSocket) + return Mono.just(createRSocket(setupPayload, sendingRSocket)) .doOnNext((rsocket) -> { - if (StringUtils.hasText(setupPayload.dataMimeType())) { - rsocket.setDataMimeType(MimeTypeUtils.parseMimeType(setupPayload.dataMimeType())); - } + String destination = rsocket.getDestination(setupPayload); Object rsocketRequesterKey = this.clientRSocketKeyStrategy.apply(destination, dataBuffer); this.clientRSocketRequesters.put(rsocketRequesterKey, rsocket.getRequester()); RSocketConnectedEvent rSocketConnectedEvent = diff --git a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/inbound/RSocketInboundGatewayIntegrationTests.java b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/inbound/RSocketInboundGatewayIntegrationTests.java index 7879847673..f77c25129c 100644 --- a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/inbound/RSocketInboundGatewayIntegrationTests.java +++ b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/inbound/RSocketInboundGatewayIntegrationTests.java @@ -47,6 +47,7 @@ import org.springframework.messaging.rsocket.RSocketRequester; import org.springframework.messaging.rsocket.RSocketStrategies; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; +import org.springframework.util.MimeType; import io.netty.buffer.PooledByteBufAllocator; import io.rsocket.frame.decoder.PayloadDecoder; @@ -218,6 +219,7 @@ public class RSocketInboundGatewayIntegrationTests { ServerRSocketConnector serverRSocketConnector = new ServerRSocketConnector(TcpServerTransport.create(tcpServer)); serverRSocketConnector.setRSocketStrategies(rsocketStrategies()); + serverRSocketConnector.setMetadataMimeType(new MimeType("message", "x.rsocket.routing.v0")); serverRSocketConnector.setFactoryConfigurer((factory) -> factory.frameDecoder(PayloadDecoder.ZERO_COPY)); return serverRSocketConnector; } @@ -231,6 +233,7 @@ public class RSocketInboundGatewayIntegrationTests { @Bean public ClientRSocketConnector clientRSocketConnector() { ClientRSocketConnector clientRSocketConnector = new ClientRSocketConnector("localhost", port); + clientRSocketConnector.setMetadataMimeType(new MimeType("message", "x.rsocket.routing.v0")); clientRSocketConnector.setFactoryConfigurer((factory) -> factory.frameDecoder(PayloadDecoder.ZERO_COPY)); clientRSocketConnector.setRSocketStrategies(rsocketStrategies()); clientRSocketConnector.setConnectRoute("clientConnect"); diff --git a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java index 3044c199dc..d47fff8bc4 100644 --- a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java +++ b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java @@ -532,6 +532,7 @@ public class RSocketOutboundGatewayIntegrationTests { return RSocketFactory.connect() .setupPayload(DefaultPayload.create("", "clientConnect")) .dataMimeType("text/plain") + .metadataMimeType("message/x.rsocket.routing.v0") .frameDecoder(PayloadDecoder.ZERO_COPY) .acceptor(clientAcceptor()) .transport(TcpClientTransport.create("localhost", port))