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

Improve RSocket support

Browse Source 
Related to: https://github.com/spring-projects/spring-boot/issues/18812

* Extract `ServerRSocketMessageHandler` into a `public` class to allow
to configure it as top-level bean and bind it into an existing server
* Change `ServerRSocketConnector` to accept the mentioned external bean
and don't create an internal RSocket server with an assumption that it
is create externally
* Add `IntegrationRSocketMessageHandler.requestMappingCompatible`
option to allow to configure `ServerRSocketMessageHandler` for both
Spring Integration RSocket channel adapters and regular `@MessageMapping`.
This is useful for Spring Boot auto-configuration

These changes give a hook to auto-configure Spring Integration RSocket
channel adapters in Spring Boot
This commit is contained in:
Artem Bilan
2019-10-31 15:39:10 -04:00
committed by Gary Russell
parent 0ceea8414f
commit 04ff879f7e
6 changed files with 248 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -23,12 +23,10 @@ import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.SmartLifecycle;
import org.springframework.lang.Nullable;
import org.springframework.messaging.rsocket.RSocketStrategies;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
import io.rsocket.metadata.WellKnownMimeType;
/**
* A base connector container for common RSocket client and server functionality.
@@ -49,18 +47,12 @@ public abstract class AbstractRSocketConnector
protected final IntegrationRSocketMessageHandler rSocketMessageHandler; // NOSONAR - final
private MimeType dataMimeType;
private MimeType metadataMimeType =
MimeTypeUtils.parseMimeType(WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.toString());
private RSocketStrategies rsocketStrategies = RSocketStrategies.create();
private boolean autoStartup = true;
private volatile boolean running;
protected AbstractRSocketConnector(IntegrationRSocketMessageHandler rSocketMessageHandler) {
Assert.notNull(rSocketMessageHandler, "'rSocketMessageHandler' must not be null");
this.rSocketMessageHandler = rSocketMessageHandler;
}
@@ -68,13 +60,13 @@ public abstract class AbstractRSocketConnector
* Configure a {@link MimeType} for data exchanging.
* @param dataMimeType the {@link MimeType} to use.
*/
public void setDataMimeType(MimeType dataMimeType) {
Assert.notNull(dataMimeType, "'dataMimeType' must not be null");
this.dataMimeType = dataMimeType;
public void setDataMimeType(@Nullable MimeType dataMimeType) {
this.rSocketMessageHandler.setDefaultDataMimeType(dataMimeType);
}
@Nullable
protected MimeType getDataMimeType() {
return this.dataMimeType;
return this.rSocketMessageHandler.getDefaultDataMimeType();
}
/**
@@ -83,12 +75,11 @@ public abstract class AbstractRSocketConnector
* @param metadataMimeType the {@link MimeType} to use.
*/
public void setMetadataMimeType(MimeType metadataMimeType) {
Assert.notNull(metadataMimeType, "'metadataMimeType' must not be null");
this.metadataMimeType = metadataMimeType;
this.rSocketMessageHandler.setDefaultMetadataMimeType(metadataMimeType);
}
protected MimeType getMetadataMimeType() {
return this.metadataMimeType;
return this.rSocketMessageHandler.getDefaultMetadataMimeType();
}
/**
@@ -96,12 +87,11 @@ public abstract class AbstractRSocketConnector
* @param rsocketStrategies the {@link RSocketStrategies} to use.
*/
public void setRSocketStrategies(RSocketStrategies rsocketStrategies) {
Assert.notNull(rsocketStrategies, "'rsocketStrategies' must not be null");
this.rsocketStrategies = rsocketStrategies;
this.rSocketMessageHandler.setRSocketStrategies(rsocketStrategies);
}
public RSocketStrategies getRSocketStrategies() {
return this.rsocketStrategies;
return this.rSocketMessageHandler.getRSocketStrategies();
}
/**
@@ -131,9 +121,6 @@ public abstract class AbstractRSocketConnector
@Override
public void afterPropertiesSet() {
this.rSocketMessageHandler.setDefaultDataMimeType(this.dataMimeType);
this.rSocketMessageHandler.setDefaultMetadataMimeType(this.metadataMimeType);
this.rSocketMessageHandler.setRSocketStrategies(this.rsocketStrategies);
this.rSocketMessageHandler.afterPropertiesSet();
}

24
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketMessageHandler.java
View File

@@ -48,8 +48,17 @@ class IntegrationRSocketMessageHandler extends RSocketMessageHandler {
private static final Method HANDLE_MESSAGE_METHOD =
ReflectionUtils.findMethod(ReactiveMessageHandler.class, "handleMessage", Message.class);
protected final boolean messageMappingCompatible; // NOSONAR final
IntegrationRSocketMessageHandler() {
setHandlerPredicate((clazz) -> false);
this(false);
}
IntegrationRSocketMessageHandler(boolean requestMappingCompatible) {
this.messageMappingCompatible = requestMappingCompatible;
if (!this.messageMappingCompatible) {
setHandlerPredicate((clazz) -> false);
}
}
public boolean detectEndpoints() {
@@ -76,14 +85,21 @@ class IntegrationRSocketMessageHandler extends RSocketMessageHandler {
@Override
protected List<? extends HandlerMethodArgumentResolver> initArgumentResolvers() {
return Collections.singletonList(new MessageHandlerMethodArgumentResolver());
if (this.messageMappingCompatible) {
// Add argument resolver before parent initializes argument resolution
getArgumentResolverConfigurer().addCustomResolver(new MessageHandlerMethodArgumentResolver());
return super.initArgumentResolvers();
}
else {
return Collections.singletonList(new MessageHandlerMethodArgumentResolver());
}
}
private static final class MessageHandlerMethodArgumentResolver implements SyncHandlerMethodArgumentResolver {
protected static final class MessageHandlerMethodArgumentResolver implements SyncHandlerMethodArgumentResolver {
@Override
public boolean supportsParameter(MethodParameter parameter) {
return true;
return Message.class.equals(parameter.getParameterType());
}
@Override

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

@@ -16,27 +16,20 @@
package org.springframework.integration.rsocket;
import java.lang.reflect.Method;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.handler.CompositeMessageCondition;
import org.springframework.messaging.handler.DestinationPatternsMessageCondition;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.messaging.rsocket.annotation.support.RSocketFrameTypeMessageCondition;
import org.springframework.messaging.rsocket.annotation.support.RSocketRequesterMethodArgumentResolver;
import org.springframework.messaging.rsocket.RSocketStrategies;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.MimeType;
import io.rsocket.RSocketFactory;
import io.rsocket.transport.ServerTransport;
@@ -56,8 +49,7 @@ import reactor.netty.http.server.HttpServer;
*
* @see RSocketFactory.ServerRSocketFactory
*/
public class ServerRSocketConnector extends AbstractRSocketConnector
implements ApplicationEventPublisherAware {
public class ServerRSocketConnector extends AbstractRSocketConnector implements ApplicationEventPublisherAware {
private final ServerTransport<CloseableChannel> serverTransport;
@@ -65,6 +57,19 @@ public class ServerRSocketConnector extends AbstractRSocketConnector
private Mono<CloseableChannel> serverMono;
/**
* Instantiate a server connector based on a provided {@link ServerRSocketMessageHandler}
* with an assumption that RSocket server is created externally as well.
* All other options are ignored in favor of provided {@link ServerRSocketMessageHandler}
* and its external RSocket server configuration.
* @param serverRSocketMessageHandler the {@link ServerRSocketMessageHandler} to rely on.
* @since 5.2.1
*/
public ServerRSocketConnector(ServerRSocketMessageHandler serverRSocketMessageHandler) {
super(serverRSocketMessageHandler);
this.serverTransport = null;
}
/**
* Instantiate a server connector based on the {@link TcpServerTransport}.
* @param bindAddress the local address to bind TCP server onto.
@@ -111,41 +116,79 @@ public class ServerRSocketConnector extends AbstractRSocketConnector
public void setClientRSocketKeyStrategy(BiFunction<Map<String, Object>,
DataBuffer, Object> clientRSocketKeyStrategy) {
Assert.notNull(clientRSocketKeyStrategy, "'clientRSocketKeyStrategy' must not be null");
serverRSocketMessageHandler().clientRSocketKeyStrategy = clientRSocketKeyStrategy;
if (this.serverTransport != null) {
serverRSocketMessageHandler().setClientRSocketKeyStrategy(clientRSocketKeyStrategy);
}
}
@Override
public void setDataMimeType(@Nullable MimeType dataMimeType) {
if (this.serverTransport != null) {
super.setDataMimeType(dataMimeType);
}
}
@Override
public void setMetadataMimeType(MimeType metadataMimeType) {
if (this.serverTransport != null) {
super.setMetadataMimeType(metadataMimeType);
}
}
@Override
public void setRSocketStrategies(RSocketStrategies rsocketStrategies) {
if (this.serverTransport != null) {
super.setRSocketStrategies(rsocketStrategies);
}
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
if (this.serverTransport != null) {
super.setApplicationContext(applicationContext);
}
}
@Override
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
serverRSocketMessageHandler().applicationEventPublisher = applicationEventPublisher;
if (this.serverTransport != null) {
serverRSocketMessageHandler().setApplicationEventPublisher(applicationEventPublisher);
}
}
@Override
public void afterPropertiesSet() {
super.afterPropertiesSet();
RSocketFactory.ServerRSocketFactory serverFactory = RSocketFactory.receive();
this.factoryConfigurer.accept(serverFactory);
if (this.serverTransport != null) {
super.afterPropertiesSet();
RSocketFactory.ServerRSocketFactory serverFactory = RSocketFactory.receive();
this.factoryConfigurer.accept(serverFactory);
this.serverMono =
serverFactory
.acceptor(serverRSocketMessageHandler().responder())
.transport(this.serverTransport)
.start()
.cache();
this.serverMono =
serverFactory
.acceptor(serverRSocketMessageHandler().responder())
.transport(this.serverTransport)
.start()
.cache();
}
}
public Map<Object, RSocketRequester> getClientRSocketRequesters() {
return Collections.unmodifiableMap(serverRSocketMessageHandler().clientRSocketRequesters);
return serverRSocketMessageHandler().getClientRSocketRequesters();
}
@Nullable
public RSocketRequester getClientRSocketRequester(Object key) {
return serverRSocketMessageHandler().clientRSocketRequesters.get(key);
return getClientRSocketRequesters().get(key);
}
public Mono<Integer> getBoundPort() {
return this.serverMono
.map((server) -> server.address().getPort());
if (this.serverTransport != null) {
return this.serverMono
.map((server) -> server.address().getPort());
}
else {
return Mono.empty();
}
}
private ServerRSocketMessageHandler serverRSocketMessageHandler() {
@@ -154,14 +197,18 @@ public class ServerRSocketConnector extends AbstractRSocketConnector
@Override
protected void doStart() {
this.serverMono.subscribe();
if (this.serverTransport != null) {
this.serverMono.subscribe();
}
}
@Override
public void destroy() {
this.serverMono
.doOnNext(Disposable::dispose)
.subscribe();
if (this.serverTransport != null) {
this.serverMono
.doOnNext(Disposable::dispose)
.subscribe();
}
}
@Override
@@ -170,47 +217,4 @@ public class ServerRSocketConnector extends AbstractRSocketConnector
serverRSocketMessageHandler().registerHandleConnectionSetupMethod();
}
private static class ServerRSocketMessageHandler extends IntegrationRSocketMessageHandler {
private static final Method HANDLE_CONNECTION_SETUP_METHOD =
ReflectionUtils.findMethod(ServerRSocketMessageHandler.class, "handleConnectionSetup", Message.class);
private final Map<Object, RSocketRequester> clientRSocketRequesters = new HashMap<>();
private BiFunction<Map<String, Object>, DataBuffer, Object> clientRSocketKeyStrategy =
(headers, data) -> data.toString(StandardCharsets.UTF_8);
private ApplicationEventPublisher applicationEventPublisher;
private void registerHandleConnectionSetupMethod() {
registerHandlerMethod(this, HANDLE_CONNECTION_SETUP_METHOD,
new CompositeMessageCondition(
RSocketFrameTypeMessageCondition.CONNECT_CONDITION,
new DestinationPatternsMessageCondition(new String[] { "*" }, obtainRouteMatcher())));
}
@SuppressWarnings("unused")
private void handleConnectionSetup(Message<DataBuffer> connectMessage) {
DataBuffer dataBuffer = connectMessage.getPayload();
MessageHeaders messageHeaders = connectMessage.getHeaders();
Object rsocketRequesterKey = this.clientRSocketKeyStrategy.apply(messageHeaders, dataBuffer);
RSocketRequester rsocketRequester =
messageHeaders.get(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER,
RSocketRequester.class);
this.clientRSocketRequesters.put(rsocketRequesterKey, rsocketRequester);
RSocketConnectedEvent rSocketConnectedEvent =
new RSocketConnectedEvent(this, messageHeaders, dataBuffer, rsocketRequester); // NOSONAR
if (this.applicationEventPublisher != null) {
this.applicationEventPublisher.publishEvent(rSocketConnectedEvent);
}
else {
if (logger.isInfoEnabled()) {
logger.info("The RSocket has been connected: " + rSocketConnectedEvent);
}
}
}
}
}

113
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketMessageHandler.java Normal file
View File

@@ -0,0 +1,113 @@
/*
* Copyright 2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.rsocket;
import java.lang.reflect.Method;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.function.BiFunction;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.handler.CompositeMessageCondition;
import org.springframework.messaging.handler.DestinationPatternsMessageCondition;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.messaging.rsocket.annotation.support.RSocketFrameTypeMessageCondition;
import org.springframework.messaging.rsocket.annotation.support.RSocketRequesterMethodArgumentResolver;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
/**
* An {@link IntegrationRSocketMessageHandler} extension for RSocket service side.
*
* @author Artem Bilan
*
* @since 5.2.1
*/
public class ServerRSocketMessageHandler extends IntegrationRSocketMessageHandler
implements ApplicationEventPublisherAware {
private static final Method HANDLE_CONNECTION_SETUP_METHOD =
ReflectionUtils.findMethod(ServerRSocketMessageHandler.class, "handleConnectionSetup", Message.class);
private final Map<Object, RSocketRequester> clientRSocketRequesters = new HashMap<>();
private BiFunction<Map<String, Object>, DataBuffer, Object> clientRSocketKeyStrategy =
(headers, data) -> data.toString(StandardCharsets.UTF_8);
private ApplicationEventPublisher applicationEventPublisher;
public ServerRSocketMessageHandler() {
this(false);
}
public ServerRSocketMessageHandler(boolean requestMappingCompatible) {
super(requestMappingCompatible);
}
public Map<Object, RSocketRequester> getClientRSocketRequesters() {
return Collections.unmodifiableMap(this.clientRSocketRequesters);
}
public void setClientRSocketKeyStrategy(
BiFunction<Map<String, Object>, DataBuffer, Object> clientRSocketKeyStrategy) {
Assert.notNull(clientRSocketKeyStrategy, "'clientRSocketKeyStrategy' must not be null");
this.clientRSocketKeyStrategy = clientRSocketKeyStrategy;
}
@Override
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
this.applicationEventPublisher = applicationEventPublisher;
}
void registerHandleConnectionSetupMethod() {
registerHandlerMethod(this, HANDLE_CONNECTION_SETUP_METHOD,
new CompositeMessageCondition(
RSocketFrameTypeMessageCondition.CONNECT_CONDITION,
new DestinationPatternsMessageCondition(new String[] { "*" }, obtainRouteMatcher())));
}
@SuppressWarnings("unused")
private void handleConnectionSetup(Message<DataBuffer> connectMessage) {
DataBuffer dataBuffer = connectMessage.getPayload();
MessageHeaders messageHeaders = connectMessage.getHeaders();
Object rsocketRequesterKey = this.clientRSocketKeyStrategy.apply(messageHeaders, dataBuffer);
RSocketRequester rsocketRequester =
messageHeaders.get(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER,
RSocketRequester.class);
this.clientRSocketRequesters.put(rsocketRequesterKey, rsocketRequester);
RSocketConnectedEvent rSocketConnectedEvent =
new RSocketConnectedEvent(this, messageHeaders, dataBuffer, rsocketRequester); // NOSONAR
if (this.applicationEventPublisher != null) {
this.applicationEventPublisher.publishEvent(rSocketConnectedEvent);
}
else {
if (logger.isInfoEnabled()) {
logger.info("The RSocket has been connected: " + rSocketConnectedEvent);
}
}
}
}

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

@@ -37,12 +37,17 @@ import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.rsocket.ClientRSocketConnector;
import org.springframework.integration.rsocket.RSocketConnectedEvent;
import org.springframework.integration.rsocket.ServerRSocketConnector;
import org.springframework.integration.rsocket.ServerRSocketMessageHandler;
import org.springframework.messaging.Message;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import io.rsocket.RSocketFactory;
import io.rsocket.frame.decoder.PayloadDecoder;
import io.rsocket.transport.netty.server.CloseableChannel;
import io.rsocket.transport.netty.server.TcpServerTransport;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
@@ -179,8 +184,23 @@ public class RSocketInboundGatewayIntegrationTests {
final MonoProcessor<RSocketRequester> clientRequester = MonoProcessor.create();
@Bean
public ServerRSocketConnector serverRSocketConnector() {
return new ServerRSocketConnector("localhost", 0);
public CloseableChannel rsocketServer() {
return RSocketFactory.receive()
.frameDecoder(PayloadDecoder.ZERO_COPY)
.acceptor(serverRSocketMessageHandler().responder())
.transport(TcpServerTransport.create("localhost", 0))
.start()
.block();
}
@Bean
public ServerRSocketMessageHandler serverRSocketMessageHandler() {
return new ServerRSocketMessageHandler(true);
}
@Bean
public ServerRSocketConnector serverRSocketConnector(ServerRSocketMessageHandler serverRSocketMessageHandler) {
return new ServerRSocketConnector(serverRSocketMessageHandler);
}
@EventListener
@@ -197,8 +217,7 @@ public class RSocketInboundGatewayIntegrationTests {
@Bean
public ClientRSocketConnector clientRSocketConnector() {
ClientRSocketConnector clientRSocketConnector =
new ClientRSocketConnector("localhost",
serverConfig.serverRSocketConnector().getBoundPort().block());
new ClientRSocketConnector("localhost", serverConfig.rsocketServer().address().getPort());
clientRSocketConnector.setSetupRoute("clientConnect/{user}");
clientRSocketConnector.setSetupRouteVariables("myUser");
return clientRSocketConnector;

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

@@ -38,6 +38,7 @@ import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.MessageChannels;
import org.springframework.integration.expression.FunctionExpression;
import org.springframework.integration.rsocket.ClientRSocketConnector;
import org.springframework.integration.rsocket.ServerRSocketMessageHandler;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
@@ -531,7 +532,7 @@ public class RSocketOutboundGatewayIntegrationTests {
@Bean
public RSocketMessageHandler messageHandler() {
return new RSocketMessageHandler();
return new ServerRSocketMessageHandler(true);
}
}