From 04ff879f7ea37c8f78aa0ee6b196b5875c27aa8e Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 31 Oct 2019 15:39:10 -0400 Subject: [PATCH] Improve RSocket support 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 --- .../rsocket/AbstractRSocketConnector.java | 33 ++-- .../IntegrationRSocketMessageHandler.java | 24 ++- .../rsocket/ServerRSocketConnector.java | 156 +++++++++--------- .../rsocket/ServerRSocketMessageHandler.java | 113 +++++++++++++ ...RSocketInboundGatewayIntegrationTests.java | 27 ++- ...SocketOutboundGatewayIntegrationTests.java | 3 +- 6 files changed, 248 insertions(+), 108 deletions(-) create mode 100644 spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketMessageHandler.java 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 e219c16431..4a88de8cd3 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 @@ -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(); } diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketMessageHandler.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketMessageHandler.java index 91f6b57244..b03fd289a6 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketMessageHandler.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketMessageHandler.java @@ -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 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 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 0034eada2c..fc0cfa94df 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 @@ -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 serverTransport; @@ -65,6 +57,19 @@ public class ServerRSocketConnector extends AbstractRSocketConnector private Mono 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, 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 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 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 clientRSocketRequesters = new HashMap<>(); - - private BiFunction, 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 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); - } - } - } - - } - } diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketMessageHandler.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketMessageHandler.java new file mode 100644 index 0000000000..a32ba61286 --- /dev/null +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketMessageHandler.java @@ -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 clientRSocketRequesters = new HashMap<>(); + + private BiFunction, 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 getClientRSocketRequesters() { + return Collections.unmodifiableMap(this.clientRSocketRequesters); + } + + public void setClientRSocketKeyStrategy( + BiFunction, 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 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); + } + } + } + +} 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 2a8df39e37..f740ec93da 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 @@ -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 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; 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 a452fcced2..0eb4e4f406 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 @@ -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); } }