diff --git a/spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistoryConfigurer.java b/spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistoryConfigurer.java index c900076b43..9631cf5067 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistoryConfigurer.java +++ b/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() + "'"); } }); 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 a2eb620934..703b6e4f08 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 @@ -35,20 +35,20 @@ import org.springframework.util.MimeTypeUtils; * A base connector container for common RSocket client and server functionality. *

* 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) { diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java index 4886c9c765..c8b014f3b9 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java @@ -37,7 +37,8 @@ import reactor.core.publisher.Mono; * A client {@link AbstractRSocketConnector} extension to the RSocket server. *

* 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 diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java index d3f9a65a9d..c1af1a0f46 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java @@ -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> 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 diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketAcceptor.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketMessageHandler.java similarity index 74% rename from spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketAcceptor.java rename to spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketMessageHandler.java index 58d4c4eeda..a160731041 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketAcceptor.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/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. *

- * The most of logic is copied from {@link org.springframework.messaging.rsocket.MessageHandlerAcceptor}. - * That cannot be extended because it is {@link final}. - *

- * 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 { +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 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 { diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/RSocketConnectedEvent.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/RSocketConnectedEvent.java index e65323af0b..a117e32774 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/RSocketConnectedEvent.java +++ b/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 { 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 558a7a9baa..a5cf00f6c4 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 @@ -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. *

* 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 serverTransport; + private final ServerTransport serverTransport; private Consumer factoryConfigurer = (serverRSocketFactory) -> { }; - private Mono serverMono; + private Mono 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 serverTransport) { - super(new ServerRSocketAcceptor()); + public ServerRSocketConnector(ServerTransport 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 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 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 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 clientRSocketRequesters = new HashMap<>(); @@ -169,32 +177,35 @@ public class ServerRSocketConnector extends AbstractRSocketConnector private ApplicationEventPublisher applicationEventPublisher; @Override - public Mono 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); + }; } } diff --git a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/dsl/RSocketDslTests.java b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/dsl/RSocketDslTests.java index 4efafc1a34..a65e679149 100644 --- a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/dsl/RSocketDslTests.java +++ b/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(); } 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 f77c25129c..d1c60a7040 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 @@ -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()); 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 a177b07fe5..fa40819ab6 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 @@ -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;