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

Fix RSocket adapters for routing metadata support

Browse Source
This commit is contained in:
Artem Bilan
2019-06-14 16:15:12 -04:00
parent 24b8e58938
commit 3ddc004a8b
7 changed files with 119 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/AbstractRSocketConnector.java
View File

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

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

@@ -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<RSocketRequester> getRSocketRequester() {
return this.rsocketMono
.map((rsocket) -> RSocketRequester.wrap(rsocket, getDataMimeType(), getRSocketStrategies()))
.map((rsocket) ->
RSocketRequester
.wrap(rsocket, getDataMimeType(), getMetadataMimeType(), getRSocketStrategies()))
.cache();
}

70
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java
View File

@@ -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<MimeType> METADATA_MIME_TYPES = Arrays.asList(COMPOSITE_METADATA, ROUTING);
private final Function<Message<?>, Mono<Void>> handler;
private final RouteMatcher routeMatcher;
private final RSocketRequester requester;
private final DataBufferFactory bufferFactory;
@Nullable
private MimeType dataMimeType;
private final MimeType dataMimeType;
IntegrationRSocket(Function<Message<?>, Mono<Void>> handler, RSocketRequester requester,
@Nullable MimeType defaultDataMimeType, DataBufferFactory bufferFactory) {
private final MimeType metadataMimeType;
IntegrationRSocket(Function<Message<?>, Mono<Void>> 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 {

48
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketAcceptor.java
View File

@@ -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<RSocket, RSocket> {
class IntegrationRSocketAcceptor extends RSocketMessageHandler
implements BiFunction<ConnectionSetupPayload, RSocket, RSocket> {
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.
* <p>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.
* <p>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 {

12
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketConnector.java
View File

@@ -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<Object, RSocketRequester> clientRSocketRequesters = new HashMap<>();
@@ -172,16 +170,12 @@ public class ServerRSocketConnector extends AbstractRSocketConnector
@Override
public Mono<RSocket> 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 =

3
spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/inbound/RSocketInboundGatewayIntegrationTests.java
View File

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

1
spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java
View File

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