Polishing RSocket module according SF changes

* Fix Checkstyle violations in the `MessageHistoryConfigurer`

spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistoryConfigurer.java

@@ -154,8 +154,8 @@ public class MessageHistoryConfigurer implements SmartLifecycle, BeanFactoryAwar
 		component.setShouldTrack(shouldTrack);
 		if (shouldTrack) {
 			this.currentlyTrackedComponents.add(component);
-			if (this.LOGGER.isInfoEnabled()) {
-				this.LOGGER.info("Enabling MessageHistory tracking for component '" + componentName + "'");
+			if (LOGGER.isInfoEnabled()) {
+				LOGGER.info("Enabling MessageHistory tracking for component '" + componentName + "'");
 			}
 		}
 	}
@@ -217,8 +217,8 @@ public class MessageHistoryConfigurer implements SmartLifecycle, BeanFactoryAwar
 			if (this.running) {
 				this.currentlyTrackedComponents.forEach(component -> {
 					component.setShouldTrack(false);
-					if (this.LOGGER.isInfoEnabled()) {
-						this.LOGGER.info("Disabling MessageHistory tracking for component '"
+					if (LOGGER.isInfoEnabled()) {
+						LOGGER.info("Disabling MessageHistory tracking for component '"
 								+ component.getComponentName() + "'");
 					}
 				});

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

@@ -35,20 +35,20 @@ import org.springframework.util.MimeTypeUtils;
  * A base connector container for common RSocket client and server functionality.
  * <p>
  * It accepts {@link IntegrationRSocketEndpoint} instances for mapping registration via an internal
- * {@link IntegrationRSocketAcceptor} or performs an auto-detection otherwise, when all bean are ready
+ * {@link IntegrationRSocketMessageHandler} or performs an auto-detection otherwise, when all bean are ready
  * in the application context.
  *
  * @author Artem Bilan
  *
  * @since 5.2
  *
- * @see IntegrationRSocketAcceptor
+ * @see IntegrationRSocketMessageHandler
  */
 public abstract class AbstractRSocketConnector
 		implements ApplicationContextAware, InitializingBean, DisposableBean, SmartInitializingSingleton,
 		SmartLifecycle {
 
-	protected final IntegrationRSocketAcceptor rsocketAcceptor; // NOSONAR - final
+	protected final IntegrationRSocketMessageHandler rSocketMessageHandler; // NOSONAR - final
 
 	private MimeType dataMimeType = MimeTypeUtils.TEXT_PLAIN;
 
@@ -65,8 +65,8 @@ public abstract class AbstractRSocketConnector
 
 	private volatile boolean running;
 
-	protected AbstractRSocketConnector(IntegrationRSocketAcceptor rsocketAcceptor) {
-		this.rsocketAcceptor = rsocketAcceptor;
+	protected AbstractRSocketConnector(IntegrationRSocketMessageHandler rSocketMessageHandler) {
+		this.rSocketMessageHandler = rSocketMessageHandler;
 	}
 
 	/**
@@ -126,25 +126,25 @@ public abstract class AbstractRSocketConnector
 	 * @param endpoint the {@link IntegrationRSocketEndpoint} to map.
 	 */
 	public void addEndpoint(IntegrationRSocketEndpoint endpoint) {
-		this.rsocketAcceptor.addEndpoint(endpoint);
+		this.rSocketMessageHandler.addEndpoint(endpoint);
 	}
 
 	@Override
 	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
-		this.rsocketAcceptor.setApplicationContext(applicationContext);
+		this.rSocketMessageHandler.setApplicationContext(applicationContext);
 	}
 
 	@Override
 	public void afterPropertiesSet() {
-		this.rsocketAcceptor.setDefaultDataMimeType(this.dataMimeType);
-		this.rsocketAcceptor.setDefaultMetadataMimeType(this.metadataMimeType);
-		this.rsocketAcceptor.setRSocketStrategies(this.rsocketStrategies);
-		this.rsocketAcceptor.afterPropertiesSet();
+		this.rSocketMessageHandler.setDefaultDataMimeType(this.dataMimeType);
+		this.rSocketMessageHandler.setDefaultMetadataMimeType(this.metadataMimeType);
+		this.rSocketMessageHandler.setRSocketStrategies(this.rsocketStrategies);
+		this.rSocketMessageHandler.afterPropertiesSet();
 	}
 
 	@Override
 	public void afterSingletonsInstantiated() {
-		this.rsocketAcceptor.detectEndpoints();
+		this.rSocketMessageHandler.detectEndpoints();
 	}
 
 	public void setAutoStartup(boolean autoStartup) {

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

@@ -37,7 +37,8 @@ import reactor.core.publisher.Mono;
  * A client {@link AbstractRSocketConnector} extension to the RSocket server.
  * <p>
  * Note: the {@link RSocketFactory.ClientRSocketFactory#acceptor(java.util.function.Function)}
- * in the provided {@link #factoryConfigurer} is overridden with an internal {@link IntegrationRSocketAcceptor}
+ * in the provided {@link #factoryConfigurer} is overridden with an internal
+ * {@link IntegrationRSocketMessageHandler#clientAcceptor()}
  * for the proper Spring Integration channel adapter mappings.
  *
  * @author Artem Bilan
@@ -85,7 +86,7 @@ public class ClientRSocketConnector extends AbstractRSocketConnector {
 	 * @param clientTransport the {@link ClientTransport} to use.
 	 */
 	public ClientRSocketConnector(ClientTransport clientTransport) {
-		super(new IntegrationRSocketAcceptor());
+		super(new IntegrationRSocketMessageHandler());
 		Assert.notNull(clientTransport, "'clientTransport' must not be null");
 		this.clientTransport = clientTransport;
 	}
@@ -125,7 +126,7 @@ public class ClientRSocketConnector extends AbstractRSocketConnector {
 						.dataMimeType(getDataMimeType().toString())
 						.metadataMimeType(getMetadataMimeType().toString());
 		this.factoryConfigurer.accept(clientFactory);
-		clientFactory.acceptor(this.rsocketAcceptor);
+		clientFactory.acceptor(this.rSocketMessageHandler.clientAcceptor());
 		Payload connectPayload = EmptyPayload.INSTANCE;
 		if (this.connectRoute != null) {
 			connectPayload = DefaultPayload.create(this.connectData, this.connectRoute);
@@ -136,7 +137,7 @@ public class ClientRSocketConnector extends AbstractRSocketConnector {
 
 	@Override
 	public void afterSingletonsInstantiated() {
-		this.autoConnect = this.rsocketAcceptor.detectEndpoints();
+		this.autoConnect = this.rSocketMessageHandler.detectEndpoints();
 	}
 
 	@Override

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

@@ -46,6 +46,7 @@ import org.springframework.util.RouteMatcher;
 
 import io.netty.buffer.ByteBuf;
 import io.rsocket.AbstractRSocket;
+import io.rsocket.ConnectionSetupPayload;
 import io.rsocket.Payload;
 import io.rsocket.metadata.CompositeMetadata;
 import reactor.core.publisher.Flux;
@@ -109,8 +110,23 @@ class IntegrationRSocket extends AbstractRSocket {
 		this.bufferFactory = bufferFactory;
 	}
 
-	public RSocketRequester getRequester() {
-		return this.requester;
+	/**
+	 * Wrap the {@link ConnectionSetupPayload} with a {@link Message} and
+	 * delegate to {@link #handle(Payload)} for handling.
+	 * @param payload the connection payload
+	 * @return completion handle for success or error
+	 */
+	Mono<Message<DataBuffer>> handleConnectionSetupPayload(ConnectionSetupPayload payload) {
+		String destination = getDestination(payload);
+		MessageHeaders headers = createHeaders(destination, null);
+		DataBuffer dataBuffer = retainDataAndReleasePayload(payload);
+		int refCount = refCount(dataBuffer);
+		return Mono.just(MessageBuilder.createMessage(dataBuffer, headers))
+				.doFinally(s -> {
+					if (refCount(dataBuffer) == refCount) {
+						DataBufferUtils.release(dataBuffer);
+					}
+				});
 	}
 
 	@Override

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

@@ -31,7 +31,6 @@ import org.springframework.messaging.handler.DestinationPatternsMessageCondition
 import org.springframework.messaging.handler.invocation.reactive.HandlerMethodArgumentResolver;
 import org.springframework.messaging.handler.invocation.reactive.SyncHandlerMethodArgumentResolver;
 import org.springframework.messaging.rsocket.RSocketRequester;
-import org.springframework.messaging.rsocket.RSocketStrategies;
 import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
 import org.springframework.util.Assert;
 import org.springframework.util.MimeType;
@@ -45,20 +44,15 @@ import io.rsocket.RSocket;
 /**
  * The {@link RSocketMessageHandler} extension for Spring Integration needs.
  * <p>
- * The most of logic is copied from {@link org.springframework.messaging.rsocket.MessageHandlerAcceptor}.
- * That cannot be extended because it is {@link final}.
- * <p>
- * This class adds an {@link IntegrationRSocketEndpoint} beans detection and registration functionality,
- * as well as serves as a container over an internal {@link IntegrationRSocket} implementation.
+ * This class adds an {@link IntegrationRSocketEndpoint} beans detection and registration functionality.
  *
  * @author Artem Bilan
  *
  * @since 5.2
  *
- * @see org.springframework.messaging.rsocket.MessageHandlerAcceptor
+ * @see RSocketMessageHandler
  */
-class IntegrationRSocketAcceptor extends RSocketMessageHandler
-		implements BiFunction<ConnectionSetupPayload, RSocket, RSocket> {
+class IntegrationRSocketMessageHandler extends RSocketMessageHandler {
 
 	private static final Method HANDLE_MESSAGE_METHOD =
 			ReflectionUtils.findMethod(ReactiveMessageHandler.class, "handleMessage", Message.class);
@@ -68,7 +62,7 @@ class IntegrationRSocketAcceptor extends RSocketMessageHandler
 
 	private MimeType defaultMetadataMimeType = IntegrationRSocket.COMPOSITE_METADATA;
 
-	IntegrationRSocketAcceptor() {
+	IntegrationRSocketMessageHandler() {
 		setHandlerPredicate((clazz) -> false);
 	}
 
@@ -84,6 +78,7 @@ class IntegrationRSocketAcceptor extends RSocketMessageHandler
 		this.defaultDataMimeType = defaultDataMimeType;
 	}
 
+
 	/**
 	 * Configure the default {@code MimeType} for payload data if the
 	 * {@code SETUP} frame did not specify one.
@@ -96,6 +91,11 @@ class IntegrationRSocketAcceptor extends RSocketMessageHandler
 		this.defaultMetadataMimeType = mimeType;
 	}
 
+	@Override
+	public BiFunction<ConnectionSetupPayload, RSocket, RSocket> clientAcceptor() {
+		return this::createRSocket;
+	}
+
 	public boolean detectEndpoints() {
 		ApplicationContext applicationContext = getApplicationContext();
 		if (applicationContext != null && getHandlerMethods().isEmpty()) {
@@ -122,27 +122,19 @@ class IntegrationRSocketAcceptor extends RSocketMessageHandler
 		return Collections.singletonList(new MessageHandlerMethodArgumentResolver());
 	}
 
-	@Override
-	public RSocket apply(ConnectionSetupPayload setupPayload, RSocket sendingRSocket) {
-		return createRSocket(setupPayload, sendingRSocket);
-	}
-
 	protected IntegrationRSocket createRSocket(ConnectionSetupPayload setupPayload, RSocket rsocket) {
-		RSocketStrategies rsocketStrategies = getRSocketStrategies();
-		MimeType dataMimeType =
-				StringUtils.hasText(setupPayload.dataMimeType())
-						? MimeTypeUtils.parseMimeType(setupPayload.dataMimeType())
-						: this.defaultDataMimeType;
-		Assert.notNull(dataMimeType, "No `dataMimeType` in the ConnectionSetupPayload and no default value");
+		String s = setupPayload.dataMimeType();
+		MimeType dataMimeType = StringUtils.hasText(s) ? MimeTypeUtils.parseMimeType(s) : this.defaultDataMimeType;
+		Assert.notNull(dataMimeType, "No `dataMimeType` in ConnectionSetupPayload and no default value");
 
-		MimeType metadataMimeType =
-				StringUtils.hasText(setupPayload.metadataMimeType())
-						? MimeTypeUtils.parseMimeType(setupPayload.metadataMimeType())
-						: this.defaultMetadataMimeType;
-		Assert.notNull(dataMimeType, "No `metadataMimeType` in the ConnectionSetupPayload and no default value");
-		return new IntegrationRSocket(this, getRouteMatcher(),
-				RSocketRequester.wrap(rsocket, dataMimeType, metadataMimeType, rsocketStrategies),
-				dataMimeType, metadataMimeType, rsocketStrategies.dataBufferFactory());
+		s = setupPayload.metadataMimeType();
+		MimeType metaMimeType = StringUtils.hasText(s) ? MimeTypeUtils.parseMimeType(s) : this.defaultMetadataMimeType;
+		Assert.notNull(dataMimeType, "No `metadataMimeType` in ConnectionSetupPayload and no default value");
+
+		RSocketRequester requester = RSocketRequester.wrap(rsocket, dataMimeType, metaMimeType, getRSocketStrategies());
+
+		return new IntegrationRSocket(this, getRouteMatcher(), requester, dataMimeType, metaMimeType,
+				getRSocketStrategies().dataBufferFactory());
 	}
 
 	private static final class MessageHandlerMethodArgumentResolver implements SyncHandlerMethodArgumentResolver {

spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/RSocketConnectedEvent.java

@@ -31,7 +31,7 @@ import org.springframework.messaging.rsocket.RSocketRequester;
  *
  * @since 5.2
  *
- * @see IntegrationRSocketAcceptor
+ * @see IntegrationRSocketMessageHandler
  */
 @SuppressWarnings("serial")
 public class RSocketConnectedEvent extends IntegrationEvent {

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

@@ -28,17 +28,18 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.core.io.buffer.DataBuffer;
-import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.lang.Nullable;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.handler.DestinationPatternsMessageCondition;
 import org.springframework.messaging.rsocket.RSocketRequester;
+import org.springframework.messaging.rsocket.annotation.support.RSocketRequesterMethodArgumentResolver;
 import org.springframework.util.Assert;
+import org.springframework.util.RouteMatcher;
 
-import io.rsocket.Closeable;
-import io.rsocket.ConnectionSetupPayload;
-import io.rsocket.RSocket;
 import io.rsocket.RSocketFactory;
 import io.rsocket.SocketAcceptor;
 import io.rsocket.transport.ServerTransport;
+import io.rsocket.transport.netty.server.CloseableChannel;
 import io.rsocket.transport.netty.server.TcpServerTransport;
 import io.rsocket.transport.netty.server.WebsocketServerTransport;
 import reactor.core.Disposable;
@@ -49,7 +50,8 @@ import reactor.netty.http.server.HttpServer;
  * A server {@link AbstractRSocketConnector} extension to accept and manage client RSocket connections.
  * <p>
  * Note: the {@link RSocketFactory.ServerRSocketFactory#acceptor(SocketAcceptor)}
- * in the provided {@link #factoryConfigurer} is overridden with an internal {@link IntegrationRSocketAcceptor}
+ * in the provided {@link #factoryConfigurer} is overridden with an internal
+ * {@link ServerRSocketMessageHandler#serverAcceptor()}
  * for the proper Spring Integration channel adapter mappings.
  *
  * @author Artem Bilan
@@ -61,11 +63,11 @@ import reactor.netty.http.server.HttpServer;
 public class ServerRSocketConnector extends AbstractRSocketConnector
 		implements ApplicationEventPublisherAware {
 
-	private final ServerTransport<? extends Closeable> serverTransport;
+	private final ServerTransport<CloseableChannel> serverTransport;
 
 	private Consumer<RSocketFactory.ServerRSocketFactory> factoryConfigurer = (serverRSocketFactory) -> { };
 
-	private Mono<? extends Closeable> serverMono;
+	private Mono<CloseableChannel> serverMono;
 
 	/**
 	 * Instantiate a server connector based on the {@link TcpServerTransport}.
@@ -90,8 +92,8 @@ public class ServerRSocketConnector extends AbstractRSocketConnector
 	 * Instantiate a server connector based on the provided {@link ServerTransport}.
 	 * @param serverTransport the {@link ServerTransport} to make server based on.
 	 */
-	public ServerRSocketConnector(ServerTransport<? extends Closeable> serverTransport) {
-		super(new ServerRSocketAcceptor());
+	public ServerRSocketConnector(ServerTransport<CloseableChannel> serverTransport) {
+		super(new ServerRSocketMessageHandler());
 		Assert.notNull(serverTransport, "'serverTransport' must not be null");
 		this.serverTransport = serverTransport;
 	}
@@ -112,12 +114,12 @@ public class ServerRSocketConnector extends AbstractRSocketConnector
 	 */
 	public void setClientRSocketKeyStrategy(BiFunction<String, DataBuffer, Object> clientRSocketKeyStrategy) {
 		Assert.notNull(clientRSocketKeyStrategy, "'clientRSocketKeyStrategy' must not be null");
-		serverRSocketAcceptor().clientRSocketKeyStrategy = clientRSocketKeyStrategy;
+		serverRSocketMessageHandler().clientRSocketKeyStrategy = clientRSocketKeyStrategy;
 	}
 
 	@Override
 	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
-		serverRSocketAcceptor().applicationEventPublisher = applicationEventPublisher;
+		serverRSocketMessageHandler().applicationEventPublisher = applicationEventPublisher;
 	}
 
 	@Override
@@ -125,25 +127,31 @@ public class ServerRSocketConnector extends AbstractRSocketConnector
 		super.afterPropertiesSet();
 		RSocketFactory.ServerRSocketFactory serverFactory = RSocketFactory.receive();
 		this.factoryConfigurer.accept(serverFactory);
+
 		this.serverMono =
 				serverFactory
-						.acceptor(serverRSocketAcceptor())
+						.acceptor(serverRSocketMessageHandler().serverAcceptor())
 						.transport(this.serverTransport)
 						.start()
 						.cache();
 	}
 
 	public Map<Object, RSocketRequester> getClientRSocketRequesters() {
-		return Collections.unmodifiableMap(serverRSocketAcceptor().clientRSocketRequesters);
+		return Collections.unmodifiableMap(serverRSocketMessageHandler().clientRSocketRequesters);
 	}
 
 	@Nullable
 	public RSocketRequester getClientRSocketRequester(Object key) {
-		return serverRSocketAcceptor().clientRSocketRequesters.get(key);
+		return serverRSocketMessageHandler().clientRSocketRequesters.get(key);
 	}
 
-	private ServerRSocketAcceptor serverRSocketAcceptor() {
-		return (ServerRSocketAcceptor) this.rsocketAcceptor;
+	public Mono<Integer> getBoundPort() {
+		return this.serverMono
+				.map((server) -> server.address().getPort());
+	}
+
+	private ServerRSocketMessageHandler serverRSocketMessageHandler() {
+		return (ServerRSocketMessageHandler) this.rSocketMessageHandler;
 	}
 
 	@Override
@@ -158,9 +166,9 @@ public class ServerRSocketConnector extends AbstractRSocketConnector
 				.subscribe();
 	}
 
-	private static class ServerRSocketAcceptor extends IntegrationRSocketAcceptor implements SocketAcceptor {
+	private static class ServerRSocketMessageHandler extends IntegrationRSocketMessageHandler {
 
-		private static final Log LOGGER = LogFactory.getLog(ServerRSocketAcceptor.class);
+		private static final Log LOGGER = LogFactory.getLog(ServerRSocketMessageHandler.class);
 
 		private final Map<Object, RSocketRequester> clientRSocketRequesters = new HashMap<>();
 
@@ -169,32 +177,35 @@ public class ServerRSocketConnector extends AbstractRSocketConnector
 		private ApplicationEventPublisher applicationEventPublisher;
 
 		@Override
-		public Mono<RSocket> accept(ConnectionSetupPayload setupPayload, RSocket sendingRSocket) {
-			DataBuffer dataBuffer =
-					IntegrationRSocket.payloadToDataBuffer(setupPayload, getRSocketStrategies().dataBufferFactory());
-			int refCount = IntegrationRSocket.refCount(dataBuffer);
-			return Mono.just(createRSocket(setupPayload, sendingRSocket))
-					.doOnNext((rsocket) -> {
-						String destination = rsocket.getDestination(setupPayload);
-						Object rsocketRequesterKey = this.clientRSocketKeyStrategy.apply(destination, dataBuffer);
-						this.clientRSocketRequesters.put(rsocketRequesterKey, rsocket.getRequester());
-						RSocketConnectedEvent rSocketConnectedEvent =
-								new RSocketConnectedEvent(rsocket, destination, dataBuffer, rsocket.getRequester());
-						if (this.applicationEventPublisher != null) {
-							this.applicationEventPublisher.publishEvent(rSocketConnectedEvent);
-						}
-						else {
-							if (LOGGER.isInfoEnabled()) {
-								LOGGER.info("The RSocket has been connected: " + rSocketConnectedEvent);
+		public SocketAcceptor serverAcceptor() {
+			return (setupPayload, sendingRSocket) -> {
+				IntegrationRSocket rsocket = createRSocket(setupPayload, sendingRSocket);
+				return rsocket.handleConnectionSetupPayload(setupPayload)
+						.doOnNext((message) -> {
+							MessageHeaders messageHeaders = message.getHeaders();
+							DataBuffer dataBuffer = message.getPayload();
+							String destination =
+									messageHeaders.get(DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER,
+											RouteMatcher.Route.class)
+											.value();
+							Object rsocketRequesterKey = this.clientRSocketKeyStrategy.apply(destination, dataBuffer);
+							RSocketRequester rsocketRequester =
+									messageHeaders.get(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER,
+											RSocketRequester.class);
+							this.clientRSocketRequesters.put(rsocketRequesterKey, rsocketRequester);
+							RSocketConnectedEvent rSocketConnectedEvent =
+									new RSocketConnectedEvent(rsocket, destination, dataBuffer, rsocketRequester);
+							if (this.applicationEventPublisher != null) {
+								this.applicationEventPublisher.publishEvent(rSocketConnectedEvent);
 							}
-						}
-					})
-					.cast(RSocket.class)
-					.doFinally((signal) -> {
-						if (IntegrationRSocket.refCount(dataBuffer) == refCount) {
-							DataBufferUtils.release(dataBuffer);
-						}
-					});
+							else {
+								if (LOGGER.isInfoEnabled()) {
+									LOGGER.info("The RSocket has been connected: " + rSocketConnectedEvent);
+								}
+							}
+						})
+						.thenReturn(rsocket);
+			};
 		}
 
 	}

spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/dsl/RSocketDslTests.java

@@ -26,7 +26,6 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.DependsOn;
 import org.springframework.core.codec.CharSequenceEncoder;
 import org.springframework.core.codec.StringDecoder;
 import org.springframework.core.io.buffer.NettyDataBufferFactory;
@@ -44,12 +43,10 @@ import io.netty.buffer.PooledByteBufAllocator;
 import io.netty.util.NetUtil;
 import io.rsocket.frame.decoder.PayloadDecoder;
 import io.rsocket.transport.netty.client.TcpClientTransport;
-import io.rsocket.transport.netty.server.TcpServerTransport;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.netty.tcp.InetSocketAddressUtil;
 import reactor.netty.tcp.TcpClient;
-import reactor.netty.tcp.TcpServer;
 
 /**
  * @author Artem Bilan
@@ -73,8 +70,6 @@ public class RSocketDslTests {
 	@EnableIntegration
 	public static class TestConfiguration {
 
-		private volatile int port;
-
 		@Bean
 		public RSocketStrategies rsocketStrategies() {
 			return RSocketStrategies.builder()
@@ -86,12 +81,7 @@ public class RSocketDslTests {
 
 		@Bean
 		public ServerRSocketConnector serverRSocketConnector() {
-			TcpServer tcpServer =
-					TcpServer.create()
-							.port(0)
-							.doOnBound((server) -> this.port = server.port());
-			ServerRSocketConnector serverRSocketConnector =
-					new ServerRSocketConnector(TcpServerTransport.create(tcpServer));
+			ServerRSocketConnector serverRSocketConnector = new ServerRSocketConnector("localhost", 0);
 			serverRSocketConnector.setRSocketStrategies(rsocketStrategies());
 			serverRSocketConnector.setFactoryConfigurer((factory) -> factory.frameDecoder(PayloadDecoder.ZERO_COPY));
 			return serverRSocketConnector;
@@ -99,15 +89,15 @@ public class RSocketDslTests {
 
 
 		@Bean
-		@DependsOn("serverRSocketConnector")
-		public ClientRSocketConnector clientRSocketConnector() {
+		public ClientRSocketConnector clientRSocketConnector(ServerRSocketConnector serverRSocketConnector) {
 			ClientRSocketConnector clientRSocketConnector =
 					new ClientRSocketConnector(
 							TcpClientTransport.create(
 									TcpClient.create()
 											.addressSupplier(() ->
 													InetSocketAddressUtil.createUnresolved(
-															NetUtil.LOCALHOST.getHostAddress(), this.port))
+															NetUtil.LOCALHOST.getHostAddress(),
+															serverRSocketConnector.getBoundPort().block()))
 							));
 			clientRSocketConnector.setFactoryConfigurer((factory) -> factory.frameDecoder(PayloadDecoder.ZERO_COPY));
 			clientRSocketConnector.setRSocketStrategies(rsocketStrategies());
@@ -116,13 +106,13 @@ public class RSocketDslTests {
 		}
 
 		@Bean
-		public IntegrationFlow rsocketUpperCaseRequestFlow() {
+		public IntegrationFlow rsocketUpperCaseRequestFlow(ClientRSocketConnector clientRSocketConnector) {
 			return IntegrationFlows
 					.from(Function.class)
 					.handle(RSockets.outboundGateway("/uppercase")
 							.command((message) -> RSocketOutboundGateway.Command.requestResponse)
 							.expectedResponseType("T(java.lang.String)")
-							.clientRSocketConnector(clientRSocketConnector()))
+							.clientRSocketConnector(clientRSocketConnector))
 					.get();
 		}
 

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

@@ -51,11 +51,9 @@ import org.springframework.util.MimeType;
 
 import io.netty.buffer.PooledByteBufAllocator;
 import io.rsocket.frame.decoder.PayloadDecoder;
-import io.rsocket.transport.netty.server.TcpServerTransport;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoProcessor;
-import reactor.netty.tcp.TcpServer;
 import reactor.test.StepVerifier;
 
 /**
@@ -69,8 +67,6 @@ public class RSocketInboundGatewayIntegrationTests {
 
 	private static AnnotationConfigApplicationContext serverContext;
 
-	private static int port;
-
 	private static ServerConfig serverConfig;
 
 	private static PollableChannel serverFireAndForgetChannelChannel;
@@ -213,11 +209,7 @@ public class RSocketInboundGatewayIntegrationTests {
 
 		@Bean
 		public ServerRSocketConnector serverRSocketConnector() {
-			TcpServer tcpServer =
-					TcpServer.create().port(0)
-							.doOnBound(server -> port = server.port());
-			ServerRSocketConnector serverRSocketConnector =
-					new ServerRSocketConnector(TcpServerTransport.create(tcpServer));
+			ServerRSocketConnector serverRSocketConnector = new ServerRSocketConnector("localhost", 0);
 			serverRSocketConnector.setRSocketStrategies(rsocketStrategies());
 			serverRSocketConnector.setMetadataMimeType(new MimeType("message", "x.rsocket.routing.v0"));
 			serverRSocketConnector.setFactoryConfigurer((factory) -> factory.frameDecoder(PayloadDecoder.ZERO_COPY));
@@ -232,7 +224,9 @@ public class RSocketInboundGatewayIntegrationTests {
 
 		@Bean
 		public ClientRSocketConnector clientRSocketConnector() {
-			ClientRSocketConnector clientRSocketConnector = new ClientRSocketConnector("localhost", port);
+			ClientRSocketConnector clientRSocketConnector =
+					new ClientRSocketConnector("localhost",
+							serverConfig.serverRSocketConnector().getBoundPort().block());
 			clientRSocketConnector.setMetadataMimeType(new MimeType("message", "x.rsocket.routing.v0"));
 			clientRSocketConnector.setFactoryConfigurer((factory) -> factory.frameDecoder(PayloadDecoder.ZERO_COPY));
 			clientRSocketConnector.setRSocketStrategies(rsocketStrategies());

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

@@ -72,7 +72,6 @@ import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoProcessor;
 import reactor.core.publisher.ReplayProcessor;
-import reactor.netty.tcp.TcpServer;
 import reactor.test.StepVerifier;
 
 /**
@@ -90,8 +89,6 @@ public class RSocketOutboundGatewayIntegrationTests {
 
 	private static AnnotationConfigApplicationContext serverContext;
 
-	private static int port;
-
 	private static CloseableChannel server;
 
 	private static FluxMessageChannel serverInputChannel;
@@ -119,13 +116,10 @@ public class RSocketOutboundGatewayIntegrationTests {
 	@BeforeAll
 	static void setup() {
 		serverContext = new AnnotationConfigApplicationContext(ServerConfig.class);
-		TcpServer tcpServer =
-				TcpServer.create().port(0)
-						.doOnBound(server -> port = server.port());
 		server = RSocketFactory.receive()
 				.frameDecoder(PayloadDecoder.ZERO_COPY)
 				.acceptor(serverContext.getBean(RSocketMessageHandler.class).serverAcceptor())
-				.transport(TcpServerTransport.create(tcpServer))
+				.transport(TcpServerTransport.create("localhost", 0))
 				.start()
 				.block();
 
@@ -537,14 +531,15 @@ public class RSocketOutboundGatewayIntegrationTests {
 					.metadataMimeType("message/x.rsocket.routing.v0")
 					.frameDecoder(PayloadDecoder.ZERO_COPY)
 					.acceptor(messageHandler().clientAcceptor())
-					.transport(TcpClientTransport.create("localhost", port))
+					.transport(TcpClientTransport.create("localhost", server.address().getPort()))
 					.start()
 					.block();
 		}
 
 		@Bean
 		public ClientRSocketConnector clientRSocketConnector() {
-			ClientRSocketConnector clientRSocketConnector = new ClientRSocketConnector("localhost", port);
+			ClientRSocketConnector clientRSocketConnector =
+					new ClientRSocketConnector("localhost", server.address().getPort());
 			clientRSocketConnector.setFactoryConfigurer((factory) -> factory.frameDecoder(PayloadDecoder.ZERO_COPY));
 			clientRSocketConnector.setRSocketStrategies(rsocketStrategies());
 			return clientRSocketConnector;