diff --git a/spring-integration-core/src/test/kotlin/org/springframework/integration/function/FunctionsTests.kt b/spring-integration-core/src/test/kotlin/org/springframework/integration/function/FunctionsTests.kt index 1acd7b8387..463484b7ae 100644 --- a/spring-integration-core/src/test/kotlin/org/springframework/integration/function/FunctionsTests.kt +++ b/spring-integration-core/src/test/kotlin/org/springframework/integration/function/FunctionsTests.kt @@ -152,14 +152,7 @@ class FunctionsTests { @Bean fun flowFromSupplier() = - IntegrationFlows.from({ "bar" }, - { e -> - e.poller { p -> - p.fixedDelay(10) - .maxMessagesPerPoll(1) - } - - }) + IntegrationFlows.from({ "bar" }) { e -> e.poller { p -> p.fixedDelay(10).maxMessagesPerPoll(1) } } .channel { c -> c.queue("fromSupplierQueue") } .get() } 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 a1bad52276..e219c16431 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 @@ -23,10 +23,6 @@ import org.springframework.beans.factory.SmartInitializingSingleton; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.context.SmartLifecycle; -import org.springframework.core.codec.CharSequenceEncoder; -import org.springframework.core.codec.StringDecoder; -import org.springframework.core.io.buffer.DefaultDataBufferFactory; -import org.springframework.messaging.rsocket.MetadataExtractor; import org.springframework.messaging.rsocket.RSocketStrategies; import org.springframework.util.Assert; import org.springframework.util.MimeType; @@ -38,7 +34,7 @@ import io.rsocket.metadata.WellKnownMimeType; * A base connector container for common RSocket client and server functionality. *

* It accepts {@link IntegrationRSocketEndpoint} instances for mapping registration via an internal - * {@link IntegrationRSocketMessageHandler} or performs an auto-detection otherwise, when all bean are ready + * {@link IntegrationRSocketMessageHandler} or performs an auto-detection otherwise, when all beans are ready * in the application context. * * @author Artem Bilan @@ -58,12 +54,7 @@ public abstract class AbstractRSocketConnector private MimeType metadataMimeType = MimeTypeUtils.parseMimeType(WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.toString()); - private RSocketStrategies rsocketStrategies = - RSocketStrategies.builder() - .decoder(StringDecoder.allMimeTypes()) - .encoder(CharSequenceEncoder.allMimeTypes()) - .dataBufferFactory(new DefaultDataBufferFactory()) - .build(); + private RSocketStrategies rsocketStrategies = RSocketStrategies.create(); private boolean autoStartup = true; @@ -114,7 +105,7 @@ public abstract class AbstractRSocketConnector } /** - * Configure {@link IntegrationRSocketEndpoint} instances for mapping nad handling requests. + * Configure {@link IntegrationRSocketEndpoint} instances for mapping and handling requests. * @param endpoints the {@link IntegrationRSocketEndpoint} instances for handling inbound requests. * @see #addEndpoint(IntegrationRSocketEndpoint) */ @@ -125,20 +116,6 @@ public abstract class AbstractRSocketConnector } } - /** - * Configure a {@link MetadataExtractor} to extract the route and possibly - * other metadata from the first payload of incoming requests. - *

By default this is a - * {@link org.springframework.messaging.rsocket.DefaultMetadataExtractor} - * with the configured {@link RSocketStrategies} (and decoders), extracting a route - * from {@code "message/x.rsocket.routing.v0"} or {@code "text/plain"} - * metadata entries. - * @param extractor the extractor to use - */ - public void setMetadataExtractor(MetadataExtractor extractor) { - this.rSocketMessageHandler.setMetadataExtractor(extractor); - } - /** * Add an {@link IntegrationRSocketEndpoint} for mapping and handling RSocket requests. * @param endpoint the {@link IntegrationRSocketEndpoint} to map. 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 a117e32774..f9709611ee 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 @@ -16,6 +16,8 @@ package org.springframework.integration.rsocket; +import java.util.Map; + import org.springframework.core.io.buffer.DataBuffer; import org.springframework.integration.events.IntegrationEvent; import org.springframework.messaging.rsocket.RSocketRequester; @@ -25,7 +27,7 @@ import org.springframework.messaging.rsocket.RSocketRequester; * to the server. *

* This event can be used for mapping {@link RSocketRequester} to the client by the - * {@code destination} meta-data or connect payload {@code data}. + * {@code headers} meta-data or connect payload {@code data}. * * @author Artem Bilan * @@ -36,21 +38,23 @@ import org.springframework.messaging.rsocket.RSocketRequester; @SuppressWarnings("serial") public class RSocketConnectedEvent extends IntegrationEvent { - private final String destination; + private final Map headers; private final DataBuffer data; private final RSocketRequester requester; - public RSocketConnectedEvent(Object source, String destination, DataBuffer data, RSocketRequester requester) { + public RSocketConnectedEvent(Object source, Map headers, DataBuffer data, + RSocketRequester requester) { + super(source); - this.destination = destination; + this.headers = headers; this.data = data; this.requester = requester; } - public String getDestination() { - return this.destination; + public Map getHeaders() { + return this.headers; } public DataBuffer getData() { @@ -64,7 +68,8 @@ public class RSocketConnectedEvent extends IntegrationEvent { @Override public String toString() { return "RSocketConnectedEvent{" + - "destination='" + this.destination + '\'' + + "headers=" + this.headers + + ", data=" + this.data + ", requester=" + this.requester + '}'; } 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 4fab5a08c1..0034eada2c 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 @@ -17,6 +17,7 @@ package org.springframework.integration.rsocket; import java.lang.reflect.Method; +import java.nio.charset.StandardCharsets; import java.util.Collections; import java.util.HashMap; import java.util.Map; @@ -36,7 +37,6 @@ import org.springframework.messaging.rsocket.annotation.support.RSocketFrameType import org.springframework.messaging.rsocket.annotation.support.RSocketRequesterMethodArgumentResolver; import org.springframework.util.Assert; import org.springframework.util.ReflectionUtils; -import org.springframework.util.RouteMatcher; import io.rsocket.RSocketFactory; import io.rsocket.transport.ServerTransport; @@ -108,7 +108,9 @@ public class ServerRSocketConnector extends AbstractRSocketConnector * 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) { + public void setClientRSocketKeyStrategy(BiFunction, + DataBuffer, Object> clientRSocketKeyStrategy) { + Assert.notNull(clientRSocketKeyStrategy, "'clientRSocketKeyStrategy' must not be null"); serverRSocketMessageHandler().clientRSocketKeyStrategy = clientRSocketKeyStrategy; } @@ -176,7 +178,8 @@ public class ServerRSocketConnector extends AbstractRSocketConnector private final Map clientRSocketRequesters = new HashMap<>(); - private BiFunction clientRSocketKeyStrategy = (destination, data) -> destination; + private BiFunction, DataBuffer, Object> clientRSocketKeyStrategy = + (headers, data) -> data.toString(StandardCharsets.UTF_8); private ApplicationEventPublisher applicationEventPublisher; @@ -184,28 +187,20 @@ public class ServerRSocketConnector extends AbstractRSocketConnector registerHandlerMethod(this, HANDLE_CONNECTION_SETUP_METHOD, new CompositeMessageCondition( RSocketFrameTypeMessageCondition.CONNECT_CONDITION, - new DestinationPatternsMessageCondition(new String[] { "*" }, getRouteMatcher()))); // NOSONAR + new DestinationPatternsMessageCondition(new String[] { "*" }, obtainRouteMatcher()))); } @SuppressWarnings("unused") private void handleConnectionSetup(Message connectMessage) { DataBuffer dataBuffer = connectMessage.getPayload(); MessageHeaders messageHeaders = connectMessage.getHeaders(); - String destination = ""; - RouteMatcher.Route route = - messageHeaders.get(DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER, - RouteMatcher.Route.class); - if (route != null) { - destination = route.value(); - } - - Object rsocketRequesterKey = this.clientRSocketKeyStrategy.apply(destination, dataBuffer); + Object rsocketRequesterKey = this.clientRSocketKeyStrategy.apply(messageHeaders, dataBuffer); RSocketRequester rsocketRequester = messageHeaders.get(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, RSocketRequester.class); this.clientRSocketRequesters.put(rsocketRequesterKey, rsocketRequester); RSocketConnectedEvent rSocketConnectedEvent = - new RSocketConnectedEvent(this, destination, dataBuffer, rsocketRequester); // NOSONAR + new RSocketConnectedEvent(this, messageHeaders, dataBuffer, rsocketRequester); // NOSONAR if (this.applicationEventPublisher != null) { this.applicationEventPublisher.publishEvent(rSocketConnectedEvent); } 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 7d169c89df..3167ca0bc9 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 @@ -22,13 +22,10 @@ import org.reactivestreams.Publisher; import org.springframework.core.ReactiveAdapter; import org.springframework.core.ResolvableType; -import org.springframework.core.codec.CharSequenceEncoder; import org.springframework.core.codec.Decoder; import org.springframework.core.codec.Encoder; -import org.springframework.core.codec.StringDecoder; import org.springframework.core.io.buffer.DataBuffer; import org.springframework.core.io.buffer.DataBufferFactory; -import org.springframework.core.io.buffer.DefaultDataBufferFactory; import org.springframework.integration.gateway.MessagingGatewaySupport; import org.springframework.integration.rsocket.AbstractRSocketConnector; import org.springframework.integration.rsocket.ClientRSocketConnector; @@ -77,12 +74,7 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In private final String[] path; - private RSocketStrategies rsocketStrategies = - RSocketStrategies.builder() - .decoder(StringDecoder.allMimeTypes()) - .encoder(CharSequenceEncoder.allMimeTypes()) - .dataBufferFactory(new DefaultDataBufferFactory()) - .build(); + private RSocketStrategies rsocketStrategies = RSocketStrategies.create(); @Nullable private AbstractRSocketConnector rsocketConnector; @@ -91,7 +83,7 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In private ResolvableType requestElementType; /** - * Instantiate based on the provided Ant-style path patterns to map this endpoint for incoming RSocket requests. + * Instantiate based on the provided path patterns to map this endpoint for incoming RSocket requests. * @param pathArg the mapping patterns to use. */ public RSocketInboundGateway(String... pathArg) { 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 770fb43b57..c387e7d82e 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,21 +26,15 @@ 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.core.codec.CharSequenceEncoder; -import org.springframework.core.codec.StringDecoder; -import org.springframework.core.io.buffer.NettyDataBufferFactory; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.rsocket.ClientRSocketConnector; import org.springframework.integration.rsocket.ServerRSocketConnector; import org.springframework.integration.rsocket.outbound.RSocketOutboundGateway; -import org.springframework.messaging.rsocket.RSocketStrategies; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; -import io.netty.buffer.PooledByteBufAllocator; -import io.rsocket.frame.decoder.PayloadDecoder; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -66,29 +60,15 @@ public class RSocketDslTests { @EnableIntegration public static class TestConfiguration { - @Bean - public RSocketStrategies rsocketStrategies() { - return RSocketStrategies.builder() - .decoder(StringDecoder.allMimeTypes()) - .encoder(CharSequenceEncoder.allMimeTypes()) - .dataBufferFactory(new NettyDataBufferFactory(PooledByteBufAllocator.DEFAULT)) - .build(); - } - @Bean public ServerRSocketConnector serverRSocketConnector() { - ServerRSocketConnector serverRSocketConnector = new ServerRSocketConnector("localhost", 0); - serverRSocketConnector.setRSocketStrategies(rsocketStrategies()); - serverRSocketConnector.setFactoryConfigurer((factory) -> factory.frameDecoder(PayloadDecoder.ZERO_COPY)); - return serverRSocketConnector; + return new ServerRSocketConnector("localhost", 0); } @Bean public ClientRSocketConnector clientRSocketConnector(ServerRSocketConnector serverRSocketConnector) { int port = serverRSocketConnector.getBoundPort().block(); ClientRSocketConnector clientRSocketConnector = new ClientRSocketConnector("localhost", port); - clientRSocketConnector.setFactoryConfigurer((factory) -> factory.frameDecoder(PayloadDecoder.ZERO_COPY)); - clientRSocketConnector.setRSocketStrategies(rsocketStrategies()); clientRSocketConnector.setAutoStartup(false); return clientRSocketConnector; } @@ -107,8 +87,7 @@ public class RSocketDslTests { @Bean public IntegrationFlow rsocketUpperCaseFlow() { return IntegrationFlows - .from(RSockets.inboundGateway("/uppercase") - .rsocketStrategies(rsocketStrategies())) + .from(RSockets.inboundGateway("/uppercase")) ., Mono>transform((flux) -> flux.next().map(String::toUpperCase)) .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 42d04eea00..2a8df39e37 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 @@ -27,15 +27,11 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestInfo; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.context.ApplicationListener; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.core.codec.CharSequenceEncoder; -import org.springframework.core.codec.StringDecoder; -import org.springframework.core.io.buffer.NettyDataBufferFactory; +import org.springframework.context.event.EventListener; import org.springframework.integration.annotation.Transformer; -import org.springframework.integration.channel.FluxMessageChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.rsocket.ClientRSocketConnector; @@ -44,13 +40,9 @@ import org.springframework.integration.rsocket.ServerRSocketConnector; import org.springframework.messaging.Message; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.rsocket.RSocketRequester; -import org.springframework.messaging.rsocket.RSocketStrategies; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; -import org.springframework.util.MimeType; -import io.netty.buffer.PooledByteBufAllocator; -import io.rsocket.frame.decoder.PayloadDecoder; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.MonoProcessor; @@ -154,15 +146,6 @@ public class RSocketInboundGatewayIntegrationTests { private abstract static class CommonConfig { - @Bean - public RSocketStrategies rsocketStrategies() { - return RSocketStrategies.builder() - .decoder(StringDecoder.allMimeTypes()) - .encoder(CharSequenceEncoder.allMimeTypes()) - .dataBufferFactory(new NettyDataBufferFactory(PooledByteBufAllocator.DEFAULT)) - .build(); - } - @Bean public PollableChannel fireAndForgetChannelChannel() { return new QueueChannel(); @@ -171,7 +154,6 @@ public class RSocketInboundGatewayIntegrationTests { @Bean public RSocketInboundGateway rsocketInboundGatewayFireAndForget() { RSocketInboundGateway rsocketInboundGateway = new RSocketInboundGateway("receive"); - rsocketInboundGateway.setRSocketStrategies(rsocketStrategies()); rsocketInboundGateway.setRequestChannel(fireAndForgetChannelChannel()); return rsocketInboundGateway; } @@ -179,16 +161,10 @@ public class RSocketInboundGatewayIntegrationTests { @Bean public RSocketInboundGateway rsocketInboundGatewayRequestReply() { RSocketInboundGateway rsocketInboundGateway = new RSocketInboundGateway("echo"); - rsocketInboundGateway.setRSocketStrategies(rsocketStrategies()); - rsocketInboundGateway.setRequestChannel(requestReplyChannel()); + rsocketInboundGateway.setRequestChannelName("requestReplyChannel"); return rsocketInboundGateway; } - @Bean - public FluxMessageChannel requestReplyChannel() { - return new FluxMessageChannel(); - } - @Transformer(inputChannel = "requestReplyChannel") public Mono echoTransformation(Flux payload) { return payload.next().map(String::toUpperCase); @@ -198,22 +174,18 @@ public class RSocketInboundGatewayIntegrationTests { @Configuration @EnableIntegration - static class ServerConfig extends CommonConfig implements ApplicationListener { + static class ServerConfig extends CommonConfig { final MonoProcessor clientRequester = MonoProcessor.create(); - @Override - public void onApplicationEvent(RSocketConnectedEvent event) { - this.clientRequester.onNext(event.getRequester()); - } - @Bean public ServerRSocketConnector serverRSocketConnector() { - 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)); - return serverRSocketConnector; + return new ServerRSocketConnector("localhost", 0); + } + + @EventListener + public void onApplicationEvent(RSocketConnectedEvent event) { + this.clientRequester.onNext(event.getRequester()); } } @@ -227,10 +199,8 @@ public class RSocketInboundGatewayIntegrationTests { 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()); - clientRSocketConnector.setSetupRoute("clientConnect"); + clientRSocketConnector.setSetupRoute("clientConnect/{user}"); + clientRSocketConnector.setSetupRouteVariables("myUser"); return clientRSocketConnector; } 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 cbe2b380f4..a452fcced2 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 @@ -30,9 +30,6 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.core.codec.CharSequenceEncoder; -import org.springframework.core.codec.StringDecoder; -import org.springframework.core.io.buffer.NettyDataBufferFactory; import org.springframework.integration.channel.FluxMessageChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; @@ -59,7 +56,6 @@ import org.springframework.stereotype.Controller; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; -import io.netty.buffer.PooledByteBufAllocator; import io.rsocket.RSocket; import io.rsocket.RSocketFactory; import io.rsocket.frame.decoder.PayloadDecoder; @@ -468,15 +464,6 @@ public class RSocketOutboundGatewayIntegrationTests { private abstract static class CommonConfig { - @Bean - public RSocketStrategies rsocketStrategies() { - return RSocketStrategies.builder() - .decoder(StringDecoder.allMimeTypes()) - .encoder(CharSequenceEncoder.allMimeTypes()) - .dataBufferFactory(new NettyDataBufferFactory(PooledByteBufAllocator.DEFAULT)) - .build(); - } - @Bean public TestController controller() { return new TestController(); @@ -515,10 +502,9 @@ public class RSocketOutboundGatewayIntegrationTests { @Bean(destroyMethod = "dispose") @Nullable public RSocket rsocketForServerRequests() { - return RSocketRequester.builder() .setupRoute("clientConnect") - .rsocketFactory(RSocketMessageHandler.clientResponder(rsocketStrategies(), controller())) + .rsocketFactory(RSocketMessageHandler.clientResponder(RSocketStrategies.create(), controller())) .connectTcp("localhost", server.address().getPort()) .block() .rsocket(); @@ -526,10 +512,7 @@ public class RSocketOutboundGatewayIntegrationTests { @Bean public ClientRSocketConnector clientRSocketConnector() { - ClientRSocketConnector clientRSocketConnector = - new ClientRSocketConnector("localhost", server.address().getPort()); - clientRSocketConnector.setRSocketStrategies(rsocketStrategies()); - return clientRSocketConnector; + return new ClientRSocketConnector("localhost", server.address().getPort()); } @Override @@ -548,9 +531,7 @@ public class RSocketOutboundGatewayIntegrationTests { @Bean public RSocketMessageHandler messageHandler() { - RSocketMessageHandler handler = new RSocketMessageHandler(); - handler.setRSocketStrategies(rsocketStrategies()); - return handler; + return new RSocketMessageHandler(); } } diff --git a/src/reference/asciidoc/dsl.adoc b/src/reference/asciidoc/dsl.adoc index 4e4f7f8002..1f1e9b01c0 100644 --- a/src/reference/asciidoc/dsl.adoc +++ b/src/reference/asciidoc/dsl.adoc @@ -1142,7 +1142,7 @@ In this case, the message handler for the first flow can be referenced with bean NOTE: An `id` attribute is required when you usE `useFlowIdAsPrefix()`. [[java-dsl-gateway]] -=== `IntegrationFlow` as Gateway +=== `IntegrationFlow` as a Gateway The `IntegrationFlow` can start from the service interface that provides a `GatewayProxyFactoryBean` component, as the following example shows: diff --git a/src/reference/asciidoc/rsocket.adoc b/src/reference/asciidoc/rsocket.adoc new file mode 100644 index 0000000000..66ece006fb --- /dev/null +++ b/src/reference/asciidoc/rsocket.adoc @@ -0,0 +1,328 @@ +[[rsocket]] +== RSocket Support + +The RSocket Spring Integration module (`spring-integration-rsocket`) allows for executions of https://rsocket.io/[RSocket application protocol]. + +You need to include this dependency into your project: + +==== +.Maven +[source, xml, subs="normal"] +---- + + org.springframework.integration + spring-integration-rsocket + {project-version} + +---- + +.Gradle +[source, groovy, subs="normal"] +---- +compile "org.springframework.integration:spring-integration-rsocket:{project-version}" +---- +==== + +This module is available starting with version 5.2 and is based on the Spring Messaging foundation with its RSocket component implementations, such as `RSocketRequester`, `RSocketMessageHandler` and `RSocketStrategies`. +See https://docs.spring.io/spring/docs/current/spring-framework-reference/web-reactive.html#rsocket[Spring Framework RSocket Support] for more information about the RSocket protocol, terminology and components. + +Before starting an integration flow processing via channel adapters, we need to establish an RSocket connection between server and client. +For this purpose, Spring Integration RSocket support provides the `ServerRSocketConnector` and `ClientRSocketConnector` implementations of the `AbstractRSocketConnector`. + +The `ServerRSocketConnector` exposes a listener on the host and port according to provided `io.rsocket.transport.ServerTransport` for accepting connections from clients. +An internal `RSocketFactory.ServerRSocketFactory` instance can be customized with the `setFactoryConfigurer()`, as well as other options that can be configured, e.g. `RSocketStrategies` and `MimeType` for payload data and headers metadata. +When a `setupRoute` is provided from the client requester (see `ClientRSocketConnector` below), a connected client is stored as a `RSocketRequester` under the key determined by the `clientRSocketKeyStrategy` `BiFunction, DataBuffer, Object>`. +By default a connect data is used for the key as a converted value to string with UTF-8 charset. +Such an `RSocketRequester` registry can be used in the application logic to determine a particular client connection for interaction with it, or for publishing the same message to all connected clients. +When a connection is established from the client, an `RSocketConnectedEvent` is emitted from the `ServerRSocketConnector`. +This is similar to what is provided by the `@ConnectMapping` annotation in Spring Messaging module. +The mapping pattern `*` means accept all the client routes. +The `RSocketConnectedEvent` can be used to distinguish different routes via `DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER` header. + +A typical server configuration might look like this: + +==== +[source, java] +---- +@Bean +public RSocketStrategies rsocketStrategies() { + return RSocketStrategies.builder() + .decoder(StringDecoder.textPlainOnly()) + .encoder(CharSequenceEncoder.allMimeTypes()) + .dataBufferFactory(new DefaultDataBufferFactory(true)) + .build(); +} + +@Bean +public ServerRSocketConnector serverRSocketConnector() { + 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)); + serverRSocketConnector.setClientRSocketKeyStrategy((headers, data) -> "" + + headers.get(DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER)); + return serverRSocketConnector; +} + +@EventListener +public void onApplicationEvent(RSocketConnectedEvent event) { + ... +} +---- +==== + +All the options, including `RSocketStrategies` bean and `@EventListener` for `RSocketConnectedEvent`, are optional. +See `ServerRSocketConnector` JavaDocs for more information. + +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. + +A typical client configuration might look like this: + +==== +[source, java] +---- +@Bean +public RSocketStrategies rsocketStrategies() { + return RSocketStrategies.builder() + .decoder(StringDecoder.textPlainOnly()) + .encoder(CharSequenceEncoder.allMimeTypes()) + .dataBufferFactory(new DefaultDataBufferFactory(true)) + .build(); +} + +@Bean +public ClientRSocketConnector clientRSocketConnector() { + ClientRSocketConnector clientRSocketConnector = + new ClientRSocketConnector("localhost", serverRSocketConnector().getBoundPort().block()); + clientRSocketConnector.setRSocketStrategies(rsocketStrategies()); + clientRSocketConnector.setSetupRoute("clientConnect/{user}"); + clientRSocketConnector.setSetupRouteVariables("myUser"); + return clientRSocketConnector; +} +---- +==== + +Most of these options (including `RSocketStrategies` bean) are optional. +Note how we connect to the locally started RSocket server on the arbitrary port. +See `ServerRSocketConnector.clientRSocketKeyStrategy` for `setupData` use cases. +Also see `ClientRSocketConnector` and its `AbstractRSocketConnector` superclass JavaDocs for more information. + +Both `ClientRSocketConnector` and `ServerRSocketConnector` are responsible for mapping inbound channel adapters to their `path` configuration for routing incoming RSocket requests. +See the next section for more information. + +[[rsocket-inbound]] +=== RSocket Inbound Gateway + +The `RSocketInboundGateway` is responsible for receiving RSocket requests and producing responses (if any). +It requires an array of `path` mapping which could be as patterns similar to MVC request mapping or `@MessageMapping` semantics. +Such a bean, according its `IntegrationRSocketEndpoint` implementation (extension of a `ReactiveMessageHandler`), is auto detected either by the `ServerRSocketConnector` or `ClientRSocketConnector` for a routing logic in the internal `IntegrationRSocketMessageHandler` for incoming requests. +An `AbstractRSocketConnector` can be provided to the `RSocketInboundGateway` for explicit endpoint registration. +This way, the auto-detection option is disabled on that `AbstractRSocketConnector`. +The `RSocketStrategies` can also be injected into the `RSocketInboundGateway` or they are obtained from the provided `AbstractRSocketConnector` overriding any explicit injection. +Decoders are used from those `RSocketStrategies` to decode a request payload according to the provided `requestElementType`. +If an `RSocketPayloadReturnValueHandler.RESPONSE_HEADER` header is not provided in incoming the `Message`, the `RSocketInboundGateway` treats a request as a `fireAndForget` RSocket interaction model. +In this case, an `RSocketInboundGateway` performs a plain `send` operation into the `outputChannel`. +Otherwise a `MonoProcessor` value from the `RSocketPayloadReturnValueHandler.RESPONSE_HEADER` header is used for sending a reply to the RSocket. +For this purpose, an `RSocketInboundGateway` performs a `sendAndReceiveMessageReactive` operation on the `outputChannel`. +The `payload` of the message to send downstream is always a `Flux` according to `MessagingRSocket` logic. +When in a `fireAndForget` RSocket interaction model, the messsage has a plain converted `payload`. +The reply `payload` could be a plain object or a `Publisher` - the `RSocketInboundGateway` converts both of them properly into an RSocket response according to the encoders provided in the `RSocketStrategies`. + +See <> for samples how to configure an `RSocketInboundGateway` endpoint and deal with payloads downstream. + +[[rsocket-outbound]] +=== RSocket Outbound Gateway + +The `RSocketOutboundGateway` is an `AbstractReplyProducingMessageHandler` to perform requests into RSocket and produce replies based on the RSocket replies (if any). +A low level RSocket protocol interaction is delegated into an `RSocketRequester` resolved from the provided `ClientRSocketConnector` or from the `RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER` header in the request message on the server side. +A target `RSocketRequester` on the server side can be resolved from an `RSocketConnectedEvent` or using `ServerRSocketConnector.getClientRSocketRequester()` API according some business key selected for connect request mappings via `ServerRSocketConnector.setClientRSocketKeyStrategy()`. +See `ServerRSocketConnector` JavaDocs for more information. + +The `route` to send request has to be configured explicitly (together with path variables) or via a SpEL expression which is evaluated against request message. + +The RSocket communication command can be provided via `RSocketOutboundGateway.Command` option or respective expression setting. +By default a `requestResponse` is used for common gateway use-cases. + +When request message payload is a `Publisher`, a `publisherElementType` option can be provided to encode its elements according an `RSocketStrategies` supplied in the target `RSocketRequester`. +An expression for this option can evaluate to a `ParameterizedTypeReference`. +See the `RSocketRequester.RequestSpec.data()` JavaDocs for more information about data and its type. + +An RSocket request can also be enhanced with a `metadata`. +For this purpose a `metadataExpression` against request message can be configured on the `RSocketOutboundGateway`. +Such an expression must evaluate to a `Map`. + +When `command` is not `fireAndForget`, an `expectedResponseType` must be supplied. +It is a `String.class` by default. +An expression for this option can evaluate to a `ParameterizedTypeReference`. +See `RSocketRequester.RequestSpec.retrieveMono()` and `RSocketRequester.RequestSpec.retrieveFlux()` JavaDocs for more information about reply data and its type. + +A reply `payload` from the `RSocketOutboundGateway` is always `Mono` (even for a `fireAndForget` command it is `Mono`) always making this component as `async`. +Such a `Mono` is subscribed before producing into the `outputChannel` for regular channels or processed on demand by the `FluxMessageChannel`. +A `Flux` response for the `requestStreamOrChannel` command is also wrapped into a mentioned reply `Mono`. +It can be flatten downstream by the `FluxMessageChannel` with a passthrough service activator: + +==== +[source, java] +---- +@ServiceActivator(inputChannel = "rsocketReplyChannel", outputChannel ="fluxMessageChannel") +public Flux flattenRSocketResponse(Flux payload) { + return payload; +} +---- +==== + +Or subscribed explicitly in the target application logic. + +See <> for samples how to configure an `RSocketOutboundGateway` endpoint a deal with payloads downstream. + +[[rsocket-namespace]] +=== RSocket Namespace Support + +Spring Integration provides an `rsocket` namespace and the corresponding schema definition. +To include it in your configuration, add the following namespace declaration in your application context configuration file: + +==== +[source,xml] +---- + + + ... + +---- +==== + +==== Inbound + +To configure Spring Integration RSocket inbound channel adapters with XML, you need to use an appropriate `inbound-gateway` components from the `int-rsocket` namespace. +The following example shows how to configure it: + +==== +[source, xml] +---- + +---- +==== + +A `ClientRSocketConnector` and `ServerRSocketConnector` should be configured as generic `` definitions. + +==== Outbound + +==== +[source, xml] +---- + +---- +==== + +See `spring-integration-rsocket.xsd` for description for all those XML attributes. + +[[rsocket-java-config]] +=== Configuring RSocket Endpoints with Java + +The following example shows how to configure an RSocket inbound endpoint with Java: + +==== +[source, java] +---- +@Bean +public RSocketInboundGateway rsocketInboundGatewayRequestReply() { + RSocketInboundGateway rsocketInboundGateway = new RSocketInboundGateway("echo"); + rsocketInboundGateway.setRequestChannelName("requestReplyChannel"); + return rsocketInboundGateway; +} + +@Transformer(inputChannel = "requestReplyChannel") +public Mono echoTransformation(Flux payload) { + return payload.next().map(String::toUpperCase); +} +---- +==== + +A `ClientRSocketConnector` or `ServerRSocketConnector` is assumed in this configuration with meaning for auto-detection of such an endpoint on the "`echo`" path. +Pay attention to the `@Transformer` signature with its fully reactive processing of the RSocket requests and producing reactive replies. + +The following example shows how to configure a RSocket inbound gateway with the Java DSL: + +==== +[source, java] +---- +@Bean +public IntegrationFlow rsocketUpperCaseFlow() { + return IntegrationFlows + .from(RSockets.inboundGateway("/uppercase")) + ., Mono>transform((flux) -> flux.next().map(String::toUpperCase)) + .get(); +} +---- +==== + +A `ClientRSocketConnector` or `ServerRSocketConnector` is assumed in this configuration with meaning for auto-detection of such an endpoint on the "`/uppercase`" path. + +The following example shows how to configure a RSocket outbound gateway with Java: + +==== +[source, java] +---- +@Bean +@ServiceActivator(inputChannel = "requestChannel", outputChannel = "replyChannel") +public RSocketOutboundGateway rsocketOutboundGateway() { + RSocketOutboundGateway rsocketOutboundGateway = + new RSocketOutboundGateway( + new FunctionExpression>((m) -> + m.getHeaders().get("route_header"))); + rsocketOutboundGateway.setCommandExpression( + new FunctionExpression>((m) -> m.getHeaders().get("rsocket_command"))); + rsocketOutboundGateway.setClientRSocketConnector(clientRSocketConnector()); + return rsocketOutboundGateway; +} +---- +==== + +The `setClientRSocketConnector()` is required only for the client side. +On the server side, the `RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER` header with an `RSocketRequester` value must be supplied in the request message. + + +The following example shows how to configure a RSocket outbound gateway with the Java DSL: + +==== +[source, java] +---- +@Bean +public IntegrationFlow rsocketUpperCaseRequestFlow(ClientRSocketConnector clientRSocketConnector) { + return IntegrationFlows + .from(Function.class) + .handle(RSockets.outboundGateway("/uppercase") + .command(RSocketOutboundGateway.Command.requestResponse) + .expectedResponseType(String.class) + .clientRSocketConnector(clientRSocketConnector)) + .get(); +} +---- +==== + +See <<./dsl.adoc#java-dsl-gateway,`IntegrationFlow` as a Gateway>> for more information how to use a mentioned `Function` interface in the beginning of the flow above. diff --git a/src/reference/asciidoc/webflux.adoc b/src/reference/asciidoc/webflux.adoc index e8ab0d1dd7..bb9e11727c 100644 --- a/src/reference/asciidoc/webflux.adoc +++ b/src/reference/asciidoc/webflux.adoc @@ -160,7 +160,7 @@ See <<./http.adoc#http-outbound,HTTP Outbound Components>> for more possible con === WebFlux Namespace Support Spring Integration provides a `webflux` namespace and the corresponding schema definition. -To include it in your configuration, include the following namespace declaration in your application context configuration file: +To include it in your configuration, add the following namespace declaration in your application context configuration file: ==== [source,xml] @@ -184,7 +184,7 @@ To include it in your configuration, include the following namespace declaration ==== Inbound -To configure Spring Integration WebFlux with XML, you caus use appropriate components from the `int-webflux` namespace: `inbound-channel-adapter` or `inbound-gateway`, corresponding to request and response requirements, respectively. +To configure Spring Integration WebFlux with XML, you need to use appropriate components from the `int-webflux` namespace: `inbound-channel-adapter` or `inbound-gateway`, corresponding to request and response requirements, respectively. The following example shows how to configure both an inbound channel adapter and an inbound gateway: ==== diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 38b8aca308..fe9cf64aa9 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -25,6 +25,12 @@ See the https://github.com/spring-projects/spring-integration/wiki/Spring-Integr [[x5.2-new-components]] === New Components +[[x5.2-rsocket-support]] +==== RSocket Support + +The `spring-integration-rsocket` module is now available with channel adapter implementations for RSocket protocol support. +See <<./rsocket.adoc#rsocket,RSocket Support>> for more information. + [[x5.2-rate-limit-advice]] ==== Rate Limit Advice Support