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 7caadf04ee..09d32af538 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 @@ -61,12 +61,14 @@ public abstract class AbstractRSocketConnector private volatile boolean running; - private ApplicationContext applicationContext; - protected AbstractRSocketConnector(IntegrationRSocketAcceptor rsocketAcceptor) { this.rsocketAcceptor = rsocketAcceptor; } + /** + * 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; @@ -76,6 +78,10 @@ public abstract class AbstractRSocketConnector return this.dataMimeType; } + /** + * Configure a {@link RSocketStrategies} for data encoding/decoding. + * @param rsocketStrategies the {@link RSocketStrategies} to use. + */ public void setRSocketStrategies(RSocketStrategies rsocketStrategies) { Assert.notNull(rsocketStrategies, "'rsocketStrategies' must not be null"); this.rsocketStrategies = rsocketStrategies; @@ -85,6 +91,11 @@ public abstract class AbstractRSocketConnector return this.rsocketStrategies; } + /** + * Configure {@link IntegrationRSocketEndpoint} instances for mapping nad handling requests. + * @param endpoints the {@link IntegrationRSocketEndpoint} instances for handling inbound requests. + * @see #addEndpoint(IntegrationRSocketEndpoint) + */ public void setEndpoints(IntegrationRSocketEndpoint... endpoints) { Assert.notNull(endpoints, "'endpoints' must not be null"); for (IntegrationRSocketEndpoint endpoint : endpoints) { @@ -92,20 +103,19 @@ public abstract class AbstractRSocketConnector } } + /** + * Add an {@link IntegrationRSocketEndpoint} for mapping and handling RSocket requests. + * @param endpoint the {@link IntegrationRSocketEndpoint} to map. + */ public void addEndpoint(IntegrationRSocketEndpoint endpoint) { this.rsocketAcceptor.addEndpoint(endpoint); } @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { - this.applicationContext = applicationContext; this.rsocketAcceptor.setApplicationContext(applicationContext); } - protected ApplicationContext getApplicationContext() { - return this.applicationContext; - } - @Override public void afterPropertiesSet() { this.rsocketAcceptor.setDefaultDataMimeType(this.dataMimeType); 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 a30af81d5f..0a9cb8e298 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 @@ -18,7 +18,6 @@ package org.springframework.integration.rsocket; import java.net.URI; import java.util.function.Consumer; -import java.util.function.Function; import org.springframework.messaging.rsocket.RSocketRequester; import org.springframework.util.Assert; @@ -37,7 +36,7 @@ import reactor.core.publisher.Mono; /** * A client {@link AbstractRSocketConnector} extension to the RSocket server. *

- * Note: the {@link RSocketFactory.ClientRSocketFactory#acceptor(Function)} + * Note: the {@link RSocketFactory.ClientRSocketFactory#acceptor(java.util.function.Function)} * in the provided {@link #factoryConfigurer} is overridden with an internal {@link IntegrationRSocketAcceptor} * for the proper Spring Integration channel adapter mappings. * @@ -62,29 +61,57 @@ public class ClientRSocketConnector extends AbstractRSocketConnector { private Mono rsocketMono; + /** + * Instantiate a connector based on the {@link TcpClientTransport}. + * @param host the TCP host to connect. + * @param port the TCP port to connect. + * @see #ClientRSocketConnector(ClientTransport) + */ public ClientRSocketConnector(String host, int port) { this(TcpClientTransport.create(host, port)); } + /** + * Instantiate a connector based on the {@link WebsocketClientTransport}. + * @param uri the WebSocket URI to connect. + * @see #ClientRSocketConnector(ClientTransport) + */ public ClientRSocketConnector(URI uri) { this(WebsocketClientTransport.create(uri)); } + /** + * Instantiate a connector based on the provided {@link ClientTransport}. + * @param clientTransport the {@link ClientTransport} to use. + */ public ClientRSocketConnector(ClientTransport clientTransport) { super(new IntegrationRSocketAcceptor()); Assert.notNull(clientTransport, "'clientTransport' must not be null"); this.clientTransport = clientTransport; } + /** + * Specify a {@link Consumer} for configuring a {@link RSocketFactory.ClientRSocketFactory}. + * @param factoryConfigurer the {@link Consumer} to configure the {@link RSocketFactory.ClientRSocketFactory}. + */ public void setFactoryConfigurer(Consumer factoryConfigurer) { Assert.notNull(factoryConfigurer, "'factoryConfigurer' must not be null"); this.factoryConfigurer = factoryConfigurer; } + /** + * Configure a route for server RSocket endpoint. + * @param connectRoute the route to connect to. + */ public void setConnectRoute(String connectRoute) { this.connectRoute = connectRoute; } + /** + * Configure a data for connect. + * Defaults to empty string. + * @param connectData the data for connect frame. + */ public void setConnectData(String connectData) { Assert.notNull(connectData, "'connectData' must not be null"); this.connectData = connectData; 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 bf218d5a63..f900f3af89 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 @@ -42,13 +42,12 @@ import org.springframework.util.MimeType; import io.netty.buffer.ByteBuf; import io.rsocket.AbstractRSocket; import io.rsocket.Payload; -import io.rsocket.RSocket; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.MonoProcessor; /** - * Implementation of {@link RSocket} that wraps incoming requests with a + * Implementation of {@link io.rsocket.RSocket} that wraps incoming requests with a * {@link Message}, delegates to a {@link Function} for handling, and then * obtains the response from a "reply" header. *

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/IntegrationRSocketAcceptor.java index 92e0e8832b..2a37bea3c1 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/IntegrationRSocketAcceptor.java @@ -37,7 +37,6 @@ import org.springframework.messaging.rsocket.RSocketStrategies; import org.springframework.util.MimeType; import org.springframework.util.ReflectionUtils; -import io.rsocket.ConnectionSetupPayload; import io.rsocket.RSocket; /** @@ -66,7 +65,7 @@ class IntegrationRSocketAcceptor extends RSocketMessageHandler implements Functi /** * Configure the default content type to use for data payloads. *

By default this is not set. However a server acceptor will use the - * content type from the {@link ConnectionSetupPayload}, so this is typically + * content type from the {@link io.rsocket.ConnectionSetupPayload}, so this is typically * required for clients but can also be used on servers as a fallback. * @param defaultDataMimeType the MimeType to use */ diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketEndpoint.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketEndpoint.java index 4cf164e6f5..1da72909b5 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketEndpoint.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketEndpoint.java @@ -33,6 +33,10 @@ import org.springframework.messaging.ReactiveMessageHandler; */ public interface IntegrationRSocketEndpoint extends ReactiveMessageHandler { + /** + * Obtain path patterns this {@link ReactiveMessageHandler} is going to be mapped onto. + * @return the path patterns for mapping. + */ String[] getPath(); } 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 4788e670c6..93a507f804 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 @@ -69,25 +69,49 @@ public class ServerRSocketConnector extends AbstractRSocketConnector private Mono serverMono; + /** + * Instantiate a server connector based on the {@link TcpServerTransport}. + * @param bindAddress the local address to bind TCP server onto. + * @param port the local TCP port to bind. + * @see #ServerRSocketConnector(ServerTransport) + */ public ServerRSocketConnector(String bindAddress, int port) { this(TcpServerTransport.create(bindAddress, port)); } + /** + * Instantiate a server connector based on the {@link WebsocketServerTransport}. + * @param server the {@link HttpServer} to use. + * @see #ServerRSocketConnector(ServerTransport) + */ public ServerRSocketConnector(HttpServer server) { this(WebsocketServerTransport.create(server)); } + /** + * 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()); Assert.notNull(serverTransport, "'serverTransport' must not be null"); this.serverTransport = serverTransport; } + /** + * Provide a {@link Consumer} to configure the {@link RSocketFactory.ServerRSocketFactory}. + * @param factoryConfigurer the {@link Consumer} to configure the {@link RSocketFactory.ServerRSocketFactory}. + */ public void setFactoryConfigurer(Consumer factoryConfigurer) { Assert.notNull(factoryConfigurer, "'factoryConfigurer' must not be null"); this.factoryConfigurer = factoryConfigurer; } + /** + * Configure a strategy to determine a key for the client {@link RSocketRequester} connected. + * Defaults to the {@code destination} the client is connected. + * @param clientRSocketKeyStrategy the {@link BiFunction} to determine a key for client {@link RSocketRequester}s. + */ public void setClientRSocketKeyStrategy(BiFunction clientRSocketKeyStrategy) { Assert.notNull(clientRSocketKeyStrategy, "'clientRSocketKeyStrategy' must not be null"); serverRSocketAcceptor().clientRSocketKeyStrategy = clientRSocketKeyStrategy; diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/RSocketNamespaceHandler.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/RSocketNamespaceHandler.java index e12159febf..a11974469d 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/RSocketNamespaceHandler.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/RSocketNamespaceHandler.java @@ -19,9 +19,11 @@ package org.springframework.integration.rsocket.config; import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler; /** - * Namespace handler for Spring Integration's RSocket namespace. + * Namespace handler for Spring Integration XML configuration for RSocket support. * * @author Artem Bilan + * + * @since 5.2 */ public class RSocketNamespaceHandler extends AbstractIntegrationNamespaceHandler { diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java index 27be78dca5..86a26de8ed 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java @@ -55,6 +55,7 @@ import reactor.core.publisher.MonoProcessor; /** * The {@link MessagingGatewaySupport} implementation for the {@link IntegrationRSocketEndpoint}. + * Represents an inbound endpoint for RSocket requests. *

* May be configured with the {@link AbstractRSocketConnector} for mapping registration. * Or existing {@link AbstractRSocketConnector} bean(s) will perform detection automatically. @@ -92,6 +93,10 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In @Nullable private ResolvableType requestElementType; + /** + * Instantiate based on the provided path patterns to map this endpoint for incoming RSocket requests. + * @param path the mapping patterns to use. + */ public RSocketInboundGateway(String... path) { Assert.notNull(path, "'path' must not be null"); this.path = path; @@ -123,7 +128,7 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In * @return the mapping path */ public String[] getPath() { - return this.path; + return Arrays.copyOf(this.path, this.path.length); } /** @@ -151,9 +156,10 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In @Override protected void onInit() { super.onInit(); - if (this.rsocketConnector != null) { - this.rsocketConnector.addEndpoint(this); - this.rsocketStrategies = this.rsocketConnector.getRSocketStrategies(); + AbstractRSocketConnector rsocketConnectorToUse = this.rsocketConnector; + if (rsocketConnectorToUse != null) { + rsocketConnectorToUse.addEndpoint(this); + this.rsocketStrategies = rsocketConnectorToUse.getRSocketStrategies(); } } diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java index 7efc5dbd70..1ac5bbc38b 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java @@ -197,8 +197,8 @@ public class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler requesterMono = this.rsocketRequesterMono; } - Assert.notNull(requesterMono, () -> - "The 'RSocketRequester' must be configured via 'ClientRSocketConnector' or provided in the '" + + Assert.notNull(requesterMono, + () -> "The 'RSocketRequester' must be configured via 'ClientRSocketConnector' or provided in the '" + RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER + "' request message headers."); return requesterMono @@ -207,13 +207,13 @@ public class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler .flatMap((responseSpec) -> performRequest(responseSpec, requestMessage)); } - private RSocketRequester.RequestSpec createRequestSpec(RSocketRequester rSocketRequester, + private RSocketRequester.RequestSpec createRequestSpec(RSocketRequester rsocketRequester, Message requestMessage) { String route = this.routeExpression.getValue(this.evaluationContext, requestMessage, String.class); Assert.notNull(route, () -> "The 'routeExpression' [" + this.routeExpression + "] must not evaluate to null"); - return rSocketRequester.route(route); + return rsocketRequester.route(route); } private RSocketRequester.ResponseSpec createResponseSpec(RSocketRequester.RequestSpec requestSpec, @@ -244,8 +244,8 @@ public class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler private Mono performRequest(RSocketRequester.ResponseSpec responseSpec, Message requestMessage) { Command command = this.commandExpression.getValue(this.evaluationContext, requestMessage, Command.class); - Assert.notNull(command, () -> "The 'commandExpression' [" + this.commandExpression + - "] must not evaluate to null"); + Assert.notNull(command, + () -> "The 'commandExpression' [" + this.commandExpression + "] must not evaluate to null"); Object expectedResponseType = null; if (!Command.fireAndForget.equals(command)) {