Polishing for RSocket channel adapter (#2928)

* Polishing for RSocket channel adapter

* Add JavaDocs
* Remove unused code

* * Fix Sonar smells
This commit is contained in:
Artem Bilan
2019-05-12 11:25:59 -04:00
committed by Gary Russell
parent a56116fc15
commit 30750f53b7
9 changed files with 95 additions and 24 deletions

View File

@@ -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);

View File

@@ -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.
* <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;

View File

@@ -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.
* <p>

View File

@@ -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.
* <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
*/

View File

@@ -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();
}

View File

@@ -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;

View File

@@ -19,9 +19,11 @@ package org.springframework.integration.rsocket.config;
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 {

View File

@@ -55,6 +55,7 @@ import reactor.core.publisher.MonoProcessor;
/**
* 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 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();
}
}

View File

@@ -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)) {