Document RSocket support

* Fix some typos in `webflux.adoc` and `dsl.adoc`
* Make `FunctionsTests.kt` more Kotlin friendly
* Improve RSocket components and tests for them,
especially rely on the default `RSocketStrategies` in the
target SF RSocket components
* Rework `ServerRSocketConnector.setClientRSocketKeyStrategy` to use the whole `MessageHeaders` for consultation.
It turns out that just destination is not enough since it can be used from different clients
* Make the key based on the provided `setupData` by default
* Include all the `MessageHeaders` into the `RSocketConnectedEvent`

* Improve Docs according review feedback and Docs in SF

Doc polishing

Author: artembilan

spring-integration-core/src/test/kotlin/org/springframework/integration/function/FunctionsTests.kt

@@ -152,14 +152,7 @@ class FunctionsTests {
 
 		@Bean
 		fun flowFromSupplier() =
-				IntegrationFlows.from<String>({ "bar" },
-						{ e ->
-							e.poller { p ->
-								p.fixedDelay(10)
-										.maxMessagesPerPoll(1)
-							}
-
-						})
+				IntegrationFlows.from<String>({ "bar" }) { e -> e.poller { p -> p.fixedDelay(10).maxMessagesPerPoll(1) } }
 						.channel { c -> c.queue("fromSupplierQueue") }
 						.get()
 	}

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

@@ -23,10 +23,6 @@
 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 @@
  * A base connector container for common RSocket client and server functionality.
  * <p>
  * 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 RSocketStrategies getRSocketStrategies() {
 	}
 
 	/**
-	 * 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 void setEndpoints(IntegrationRSocketEndpoint... endpoints) {
 		}
 	}
 
-	/**
-	 * Configure a {@link MetadataExtractor} to extract the route and possibly
-	 * other metadata from the first payload of incoming requests.
-	 * <p>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.

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 @@
  * to the server.
  * <p>
  * 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 @@
 @SuppressWarnings("serial")
 public class RSocketConnectedEvent extends IntegrationEvent {
 
-	private final String destination;
+	private final Map<String, Object> headers;
 
 	private final DataBuffer data;
 
 	private final RSocketRequester requester;
 
-	public RSocketConnectedEvent(Object source, String destination, DataBuffer data, RSocketRequester requester) {
+	public RSocketConnectedEvent(Object source, Map<String, Object> 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<String, Object> getHeaders() {
+		return this.headers;
 	}
 
 	public DataBuffer getData() {
@@ -64,7 +68,8 @@ public RSocketRequester getRequester() {
 	@Override
 	public String toString() {
 		return "RSocketConnectedEvent{" +
-				"destination='" + this.destination + '\'' +
+				"headers=" + this.headers +
+				", data=" + this.data +
 				", requester=" + this.requester +
 				'}';
 	}

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.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 void setFactoryConfigurer(Consumer<RSocketFactory.ServerRSocketFactory> f
 	 * 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) {
+	public void setClientRSocketKeyStrategy(BiFunction<Map<String, Object>,
+			DataBuffer, Object> clientRSocketKeyStrategy) {
+
 		Assert.notNull(clientRSocketKeyStrategy, "'clientRSocketKeyStrategy' must not be null");
 		serverRSocketMessageHandler().clientRSocketKeyStrategy = clientRSocketKeyStrategy;
 	}
@@ -176,36 +178,29 @@ private static class ServerRSocketMessageHandler extends IntegrationRSocketMessa
 
 		private final Map<Object, RSocketRequester> clientRSocketRequesters = new HashMap<>();
 
-		private BiFunction<String, DataBuffer, Object> clientRSocketKeyStrategy = (destination, data) -> destination;
+		private BiFunction<Map<String, Object>, DataBuffer, Object> clientRSocketKeyStrategy =
+				(headers, data) -> data.toString(StandardCharsets.UTF_8);
 
 		private ApplicationEventPublisher applicationEventPublisher;
 
 		private void registerHandleConnectionSetupMethod() {
 			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<DataBuffer> 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);
 			}

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

@@ -22,13 +22,10 @@
 
 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) {

spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/dsl/RSocketDslTests.java

@@ -26,21 +26,15 @@
 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 @@ void testRsocketUpperCaseFlows() {
 	@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 IntegrationFlow rsocketUpperCaseRequestFlow(ClientRSocketConnector client
 		@Bean
 		public IntegrationFlow rsocketUpperCaseFlow() {
 			return IntegrationFlows
-					.from(RSockets.inboundGateway("/uppercase")
-							.rsocketStrategies(rsocketStrategies()))
+					.from(RSockets.inboundGateway("/uppercase"))
 					.<Flux<String>, Mono<String>>transform((flux) -> flux.next().map(String::toUpperCase))
 					.get();
 		}