diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketMessageHandler.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketMessageHandler.java index b03fd289a6..a0fa75ffac 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketMessageHandler.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketMessageHandler.java @@ -54,8 +54,8 @@ class IntegrationRSocketMessageHandler extends RSocketMessageHandler { this(false); } - IntegrationRSocketMessageHandler(boolean requestMappingCompatible) { - this.messageMappingCompatible = requestMappingCompatible; + IntegrationRSocketMessageHandler(boolean messageMappingCompatible) { + this.messageMappingCompatible = messageMappingCompatible; if (!this.messageMappingCompatible) { setHandlerPredicate((clazz) -> false); } 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 fc0cfa94df..20640195ab 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 @@ -99,6 +99,10 @@ public class ServerRSocketConnector extends AbstractRSocketConnector implements this.serverTransport = serverTransport; } + private ServerRSocketMessageHandler serverRSocketMessageHandler() { + return (ServerRSocketMessageHandler) this.rSocketMessageHandler; + } + /** * Provide a {@link Consumer} to configure the {@link RSocketFactory.ServerRSocketFactory}. * @param factoryConfigurer the {@link Consumer} to configure the {@link RSocketFactory.ServerRSocketFactory}. @@ -110,7 +114,7 @@ public class ServerRSocketConnector extends AbstractRSocketConnector implements /** * Configure a strategy to determine a key for the client {@link RSocketRequester} connected. - * Defaults to the {@code destination} the client is connected. + * Defaults to the {@code destination} to which a client is connected. * @param clientRSocketKeyStrategy the {@link BiFunction} to determine a key for client {@link RSocketRequester}s. */ public void setClientRSocketKeyStrategy(BiFunction, @@ -172,15 +176,31 @@ public class ServerRSocketConnector extends AbstractRSocketConnector implements } } + /** + * Return connected {@link RSocketRequester}s mapped by keys. + * @return connected {@link RSocketRequester}s mapped by keys. + * @see ServerRSocketMessageHandler#getClientRSocketRequesters() + */ public Map getClientRSocketRequesters() { return serverRSocketMessageHandler().getClientRSocketRequesters(); } + /** + * Return connected {@link RSocketRequester} mapped by key or null. + * @param key the mapping key. + * @return the {@link RSocketRequester} or null. + * @see ServerRSocketMessageHandler#getClientRSocketRequester(Object) + */ @Nullable public RSocketRequester getClientRSocketRequester(Object key) { - return getClientRSocketRequesters().get(key); + return serverRSocketMessageHandler().getClientRSocketRequester(key); } + /** + * Return the port this internal server is bound or empty {@link Mono}. + * @return the port this internal server is bound or empty {@link Mono} + * if an external server is used. + */ public Mono getBoundPort() { if (this.serverTransport != null) { return this.serverMono @@ -191,10 +211,6 @@ public class ServerRSocketConnector extends AbstractRSocketConnector implements } } - private ServerRSocketMessageHandler serverRSocketMessageHandler() { - return (ServerRSocketMessageHandler) this.rSocketMessageHandler; - } - @Override protected void doStart() { if (this.serverTransport != null) { diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketMessageHandler.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketMessageHandler.java index a32ba61286..35e7a40635 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketMessageHandler.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketMessageHandler.java @@ -26,6 +26,7 @@ import java.util.function.BiFunction; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.handler.CompositeMessageCondition; @@ -38,6 +39,18 @@ import org.springframework.util.ReflectionUtils; /** * An {@link IntegrationRSocketMessageHandler} extension for RSocket service side. + *

+ * In a plain Spring Integration application instances of this class are created by the + * {@link ServerRSocketConnector} internally and a new RSocket server is started over there. + * When an existing RSocket server is in use, an instance of this class has to be + * provided as a {@link #responder()} into that server and a {@link ServerRSocketConnector} + * should accept the same instance as a delegate. + *

+ * With a {@link #messageMappingCompatible} option this class also handles + * {@link org.springframework.messaging.handler.annotation.MessageMapping} methods, + * covering both Spring Integration and standard + * {@link org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler} + * functionality. * * @author Artem Bilan * @@ -57,19 +70,35 @@ public class ServerRSocketMessageHandler extends IntegrationRSocketMessageHandle private ApplicationEventPublisher applicationEventPublisher; + /** + * Create an service side RSocket message handler instance for delegating + * to {@link IntegrationRSocketEndpoint} beans and collect {@link RSocketRequester}s + * from client connections. + */ public ServerRSocketMessageHandler() { this(false); } - public ServerRSocketMessageHandler(boolean requestMappingCompatible) { - super(requestMappingCompatible); - } - - - public Map getClientRSocketRequesters() { - return Collections.unmodifiableMap(this.clientRSocketRequesters); + /** + * Create an service side RSocket message handler instance for delegating + * to {@link IntegrationRSocketEndpoint} beans and collect {@link RSocketRequester}s + * from client connections. + * When {@code messageMappingCompatible == true}, this class also handles + * {@link org.springframework.messaging.handler.annotation.MessageMapping} methods + * as it is done by the standard + * {@link org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler}. + * @param messageMappingCompatible whether handle also + * {@link org.springframework.messaging.handler.annotation.MessageMapping}. + */ + public ServerRSocketMessageHandler(boolean messageMappingCompatible) { + super(messageMappingCompatible); } + /** + * Configure a {@link BiFunction} to extract a key for mapping connected {@link RSocketRequester}s. + * Defaults to the {@code destination} a client is connected. + * @param clientRSocketKeyStrategy the {@link BiFunction} to use. + */ public void setClientRSocketKeyStrategy( BiFunction, DataBuffer, Object> clientRSocketKeyStrategy) { @@ -77,6 +106,25 @@ public class ServerRSocketMessageHandler extends IntegrationRSocketMessageHandle this.clientRSocketKeyStrategy = clientRSocketKeyStrategy; } + /** + * Get connected {@link RSocketRequester}s mapped by the keys from the connect messages. + * @return the map of connected {@link RSocketRequester}s. + * @see #setClientRSocketKeyStrategy + */ + public Map getClientRSocketRequesters() { + return Collections.unmodifiableMap(this.clientRSocketRequesters); + } + + /** + * Obtain a connected {@link RSocketRequester} mapped by provided key or null. + * @param key the key for mapped {@link RSocketRequester} if any. + * @return the mapped {@link RSocketRequester} or null. + */ + @Nullable + public RSocketRequester getClientRSocketRequester(Object key) { + return this.clientRSocketRequesters.get(key); + } + @Override public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { this.applicationEventPublisher = applicationEventPublisher; diff --git a/src/reference/asciidoc/rsocket.adoc b/src/reference/asciidoc/rsocket.adoc index daf161ad77..49214fb7e7 100644 --- a/src/reference/asciidoc/rsocket.adoc +++ b/src/reference/asciidoc/rsocket.adoc @@ -74,6 +74,11 @@ public void onApplicationEvent(RSocketConnectedEvent event) { All the options, including `RSocketStrategies` bean and `@EventListener` for `RSocketConnectedEvent`, are optional. See `ServerRSocketConnector` JavaDocs for more information. +Starting with version 5.2.1, the `ServerRSocketMessageHandler` is extracted to a public, top-level class for possible connection with an existing RSocket server. +When a `ServerRSocketConnector` is supplied with an external instance of `ServerRSocketMessageHandler`, it doesn't create an RSocket server internally and just delegates all the handling logic to the provided instance. +In addition the `ServerRSocketMessageHandler` can be configured with a `messageMappingCompatible` flag to handle also `@MessageMapping` for an RSocket controller, fully replacing the functionality provided by the standard `RSocketMessageHandler`. +This can be useful in mixed configurations, when classic `@MessageMapping` methods are present in the same application along with RSocket channel adapters and an externally configured RSocket server is present in the application. + The `ClientRSocketConnector` serves as a holder for `RSocketRequester` based on the `RSocket` connected via the provided `ClientTransport`. The `RSocketFactory.ClientRSocketFactory` can be customized with the provided `ClientRSocketFactoryConfigurer`. The `setupRoute` (with optional templates variables) and `setupData` with metadata can be also configured on this component.