Polishing for RSocket channel adapter (#2928)

* Polishing for RSocket channel adapter

* Add JavaDocs
* Remove unused code

* * Fix Sonar smells

Author: artembilan

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 @@ protected MimeType getDataMimeType() {
 		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,27 +91,31 @@ public RSocketStrategies getRSocketStrategies() {
 		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) {
 			addEndpoint(endpoint);
 		}
 	}
 
+	/**
+	 * 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);

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

@@ -18,7 +18,6 @@
 
 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 @@
 /**
  * A client {@link AbstractRSocketConnector} extension to the RSocket server.
  * <p>
- * 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<RSocket> 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<RSocketFactory.ClientRSocketFactory> 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;

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

@@ -42,13 +42,12 @@
 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.
  * <p>

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

@@ -37,7 +37,6 @@
 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.
 	 * <p>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
 	 */

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

@@ -33,6 +33,10 @@
  */
 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();
 
 }

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

@@ -69,25 +69,49 @@ public class ServerRSocketConnector extends AbstractRSocketConnector
 
 	private Mono<? extends Closeable> 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<? extends Closeable> 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<RSocketFactory.ServerRSocketFactory> 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<String, DataBuffer, Object> clientRSocketKeyStrategy) {
 		Assert.notNull(clientRSocketKeyStrategy, "'clientRSocketKeyStrategy' must not be null");
 		serverRSocketAcceptor().clientRSocketKeyStrategy = clientRSocketKeyStrategy;

spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/RSocketNamespaceHandler.java

@@ -19,9 +19,11 @@
 import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler;
 
 /**
- * Namespace handler for Spring Integration's <em>RSocket</em> namespace.
+ * Namespace handler for Spring Integration XML configuration for <em>RSocket</em> support.
  *
  * @author Artem Bilan
+ *
+ * @since 5.2
  */
 public class RSocketNamespaceHandler extends AbstractIntegrationNamespaceHandler {
 

spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java

@@ -55,6 +55,7 @@
 
 /**
  * The {@link MessagingGatewaySupport} implementation for the {@link IntegrationRSocketEndpoint}.
+ * Represents an inbound endpoint for RSocket requests.
  * <p>
  * 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 void setRSocketConnector(AbstractRSocketConnector rsocketConnector) {
 	 * @return the mapping path
 	 */
 	public String[] getPath() {
-		return this.path;
+		return Arrays.copyOf(this.path, this.path.length);
 	}
 
 	/**
@@ -151,9 +156,10 @@ public void setRequestElementType(ResolvableType requestElementType) {
 	@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();
 		}
 	}
 

spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java

@@ -197,8 +197,8 @@ protected Object handleRequestMessage(Message<?> requestMessage) {
 			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 @@ protected Object handleRequestMessage(Message<?> requestMessage) {
 				.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 @@ private RSocketRequester.ResponseSpec responseSpecForPublisher(RSocketRequester.
 
 	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)) {