Merge branch 'rs'

Author: rstoyanchev

spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/reactive/MessageMappingMessageHandler.java

@@ -88,6 +88,7 @@ public class MessageMappingMessageHandler extends AbstractMethodMessageHandler<C
 	@Nullable
 	private Validator validator;
 
+	@Nullable
 	private RouteMatcher routeMatcher;
 
 	private ConversionService conversionService = new DefaultFormattingConversionService();
@@ -97,9 +98,6 @@ public class MessageMappingMessageHandler extends AbstractMethodMessageHandler<C
 
 
 	public MessageMappingMessageHandler() {
-		AntPathMatcher pathMatcher = new AntPathMatcher();
-		pathMatcher.setPathSeparator(".");
-		this.routeMatcher = new SimpleRouteMatcher(pathMatcher);
 		setHandlerPredicate(type -> AnnotatedElementUtils.hasAnnotation(type, Controller.class));
 	}
 
@@ -150,7 +148,9 @@ public void setRouteMatcher(RouteMatcher routeMatcher) {
 
 	/**
 	 * Return the {@code RouteMatcher} used to map messages to handlers.
+	 * May be {@code null} before component is initialized.
 	 */
+	@Nullable
 	public RouteMatcher getRouteMatcher() {
 		return this.routeMatcher;
 	}
@@ -203,6 +203,12 @@ protected List<? extends HandlerMethodArgumentResolver> initArgumentResolvers()
 		resolvers.add(new PayloadMethodArgumentResolver(
 				getDecoders(), this.validator, getReactiveAdapterRegistry(), true));
 
+		if (this.routeMatcher == null) {
+			AntPathMatcher pathMatcher = new AntPathMatcher();
+			pathMatcher.setPathSeparator(".");
+			this.routeMatcher = new SimpleRouteMatcher(pathMatcher);
+		}
+
 		return resolvers;
 	}
 

spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequesterBuilder.java

@@ -18,9 +18,9 @@
 
 import java.net.URI;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.function.Consumer;
-import java.util.stream.Stream;
 
 import io.rsocket.RSocketFactory;
 import io.rsocket.frame.decoder.PayloadDecoder;
@@ -29,6 +29,9 @@
 import io.rsocket.transport.netty.client.WebsocketClientTransport;
 import reactor.core.publisher.Mono;
 
+import org.springframework.core.codec.Decoder;
+import org.springframework.core.codec.StringDecoder;
+import org.springframework.core.io.buffer.NettyDataBufferFactory;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.MimeType;
@@ -110,7 +113,10 @@ private Mono<RSocketRequester> doConnect(ClientTransport transport) {
 		MimeType dataMimeType = getDataMimeType(rsocketStrategies);
 		rsocketFactory.dataMimeType(dataMimeType.toString());
 		rsocketFactory.metadataMimeType(this.metadataMimeType.toString());
-		rsocketFactory.frameDecoder(PayloadDecoder.ZERO_COPY);
+
+		if (rsocketStrategies.dataBufferFactory() instanceof NettyDataBufferFactory) {
+			rsocketFactory.frameDecoder(PayloadDecoder.ZERO_COPY);
+		}
 
 		this.rsocketFactoryConfigurers.forEach(configurer -> {
 			configurer.configureWithStrategies(rsocketStrategies);
@@ -139,16 +145,35 @@ private MimeType getDataMimeType(RSocketStrategies strategies) {
 		if (this.dataMimeType != null) {
 			return this.dataMimeType;
 		}
-		return Stream
-				.concat(
-						strategies.encoders().stream()
-								.flatMap(encoder -> encoder.getEncodableMimeTypes().stream()),
-						strategies.decoders().stream()
-								.flatMap(encoder -> encoder.getDecodableMimeTypes().stream())
-				)
-				.filter(MimeType::isConcrete)
-				.findFirst()
-				.orElseThrow(() -> new IllegalArgumentException("Failed to select data MimeType to use."));
+		// Look for non-basic Decoder (e.g. CBOR, Protobuf)
+		MimeType selected = null;
+		List<Decoder<?>> decoders = strategies.decoders();
+		for (Decoder<?> candidate : decoders) {
+			if (!isCoreCodec(candidate) && !candidate.getDecodableMimeTypes().isEmpty()) {
+				Assert.state(selected == null,
+						() -> "Cannot select default data MimeType based on configured decoders: " + decoders);
+				selected = getMimeType(candidate);
+			}
+		}
+		if (selected != null) {
+			return selected;
+		}
+		// Fall back on 1st decoder (e.g. String)
+		for (Decoder<?> decoder : decoders) {
+			if (!decoder.getDecodableMimeTypes().isEmpty()) {
+				return getMimeType(decoder);
+			}
+		}
+		throw new IllegalArgumentException("Failed to select data MimeType to use.");
+	}
+
+	private static boolean isCoreCodec(Object codec) {
+		return codec.getClass().getPackage().equals(StringDecoder.class.getPackage());
+	}
+
+	private static MimeType getMimeType(Decoder<?> decoder) {
+		MimeType mimeType = decoder.getDecodableMimeTypes().get(0);
+		return new MimeType(mimeType, Collections.emptyMap());
 	}
 
 }

spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketStrategies.java

@@ -22,13 +22,27 @@
 import java.util.List;
 import java.util.function.Consumer;
 
+import io.netty.buffer.PooledByteBufAllocator;
+
 import org.springframework.core.ReactiveAdapterRegistry;
+import org.springframework.core.codec.ByteArrayDecoder;
+import org.springframework.core.codec.ByteArrayEncoder;
+import org.springframework.core.codec.ByteBufferDecoder;
+import org.springframework.core.codec.ByteBufferEncoder;
+import org.springframework.core.codec.CharSequenceEncoder;
+import org.springframework.core.codec.DataBufferDecoder;
+import org.springframework.core.codec.DataBufferEncoder;
 import org.springframework.core.codec.Decoder;
 import org.springframework.core.codec.Encoder;
+import org.springframework.core.codec.StringDecoder;
 import org.springframework.core.io.buffer.DataBufferFactory;
-import org.springframework.core.io.buffer.DefaultDataBufferFactory;
+import org.springframework.core.io.buffer.NettyDataBufferFactory;
 import org.springframework.lang.Nullable;
+import org.springframework.util.AntPathMatcher;
 import org.springframework.util.Assert;
+import org.springframework.util.MimeTypeUtils;
+import org.springframework.util.RouteMatcher;
+import org.springframework.util.SimpleRouteMatcher;
 
 /**
  * Default, package-private {@link RSocketStrategies} implementation.
@@ -42,18 +56,25 @@ final class DefaultRSocketStrategies implements RSocketStrategies {
 
 	private final List<Decoder<?>> decoders;
 
-	private final ReactiveAdapterRegistry adapterRegistry;
+	private final RouteMatcher routeMatcher;
+
+	private final MetadataExtractor metadataExtractor;
 
 	private final DataBufferFactory bufferFactory;
 
+	private final ReactiveAdapterRegistry adapterRegistry;
+
 
 	private DefaultRSocketStrategies(List<Encoder<?>> encoders, List<Decoder<?>> decoders,
-			ReactiveAdapterRegistry adapterRegistry, DataBufferFactory bufferFactory) {
+			RouteMatcher routeMatcher, MetadataExtractor metadataExtractor,
+			DataBufferFactory bufferFactory, ReactiveAdapterRegistry adapterRegistry) {
 
 		this.encoders = Collections.unmodifiableList(encoders);
 		this.decoders = Collections.unmodifiableList(decoders);
-		this.adapterRegistry = adapterRegistry;
+		this.routeMatcher = routeMatcher;
+		this.metadataExtractor = metadataExtractor;
 		this.bufferFactory = bufferFactory;
+		this.adapterRegistry = adapterRegistry;
 	}
 
 
@@ -68,15 +89,25 @@ public List<Decoder<?>> decoders() {
 	}
 
 	@Override
-	public ReactiveAdapterRegistry reactiveAdapterRegistry() {
-		return this.adapterRegistry;
+	public RouteMatcher routeMatcher() {
+		return this.routeMatcher;
+	}
+
+	@Override
+	public MetadataExtractor metadataExtractor() {
+		return this.metadataExtractor;
 	}
 
 	@Override
 	public DataBufferFactory dataBufferFactory() {
 		return this.bufferFactory;
 	}
 
+	@Override
+	public ReactiveAdapterRegistry reactiveAdapterRegistry() {
+		return this.adapterRegistry;
+	}
+
 
 	/**
 	 * Default RSocketStrategies.Builder implementation.
@@ -87,21 +118,42 @@ static class DefaultRSocketStrategiesBuilder implements RSocketStrategies.Builde
 
 		private final List<Decoder<?>> decoders = new ArrayList<>();
 
+		@Nullable
+		private RouteMatcher routeMatcher;
+
+		@Nullable
+		private MetadataExtractor metadataExtractor;
+
 		private ReactiveAdapterRegistry adapterRegistry = ReactiveAdapterRegistry.getSharedInstance();
 
 		@Nullable
-		private DataBufferFactory dataBufferFactory;
+		private DataBufferFactory bufferFactory;
+
+
+		DefaultRSocketStrategiesBuilder() {
+
+			// Order of decoders may be significant for default data MimeType
+			// selection in RSocketRequester.Builder
+
+			this.decoders.add(StringDecoder.allMimeTypes());
+			this.decoders.add(new ByteBufferDecoder());
+			this.decoders.add(new ByteArrayDecoder());
+			this.decoders.add(new DataBufferDecoder());
 
-		public DefaultRSocketStrategiesBuilder() {
+			this.encoders.add(CharSequenceEncoder.allMimeTypes());
+			this.encoders.add(new ByteBufferEncoder());
+			this.encoders.add(new ByteArrayEncoder());
+			this.encoders.add(new DataBufferEncoder());
 		}
 
-		public DefaultRSocketStrategiesBuilder(RSocketStrategies other) {
+		DefaultRSocketStrategiesBuilder(RSocketStrategies other) {
 			this.encoders.addAll(other.encoders());
 			this.decoders.addAll(other.decoders());
 			this.adapterRegistry = other.reactiveAdapterRegistry();
-			this.dataBufferFactory = other.dataBufferFactory();
+			this.bufferFactory = other.dataBufferFactory();
 		}
 
+
 		@Override
 		public Builder encoder(Encoder<?>... encoders) {
 			this.encoders.addAll(Arrays.asList(encoders));
@@ -126,6 +178,18 @@ public Builder decoders(Consumer<List<Decoder<?>>> consumer) {
 			return this;
 		}
 
+		@Override
+		public Builder routeMatcher(RouteMatcher routeMatcher) {
+			this.routeMatcher = routeMatcher;
+			return this;
+		}
+
+		@Override
+		public Builder metadataExtractor(MetadataExtractor metadataExtractor) {
+			this.metadataExtractor = metadataExtractor;
+			return this;
+		}
+
 		@Override
 		public Builder reactiveAdapterStrategy(ReactiveAdapterRegistry registry) {
 			Assert.notNull(registry, "ReactiveAdapterRegistry is required");
@@ -135,14 +199,34 @@ public Builder reactiveAdapterStrategy(ReactiveAdapterRegistry registry) {
 
 		@Override
 		public Builder dataBufferFactory(DataBufferFactory bufferFactory) {
-			this.dataBufferFactory = bufferFactory;
+			this.bufferFactory = bufferFactory;
 			return this;
 		}
 
 		@Override
 		public RSocketStrategies build() {
-			return new DefaultRSocketStrategies(this.encoders, this.decoders, this.adapterRegistry,
-					this.dataBufferFactory != null ? this.dataBufferFactory : new DefaultDataBufferFactory());
+			return new DefaultRSocketStrategies(
+					this.encoders, this.decoders,
+					this.routeMatcher != null ? this.routeMatcher : initRouteMatcher(),
+					this.metadataExtractor != null ? this.metadataExtractor : initMetadataExtractor(),
+					this.bufferFactory != null ? this.bufferFactory : initBufferFactory(),
+					this.adapterRegistry);
+		}
+
+		private RouteMatcher initRouteMatcher() {
+			AntPathMatcher pathMatcher = new AntPathMatcher();
+			pathMatcher.setPathSeparator(".");
+			return new SimpleRouteMatcher(pathMatcher);
+		}
+
+		private MetadataExtractor initMetadataExtractor() {
+			DefaultMetadataExtractor extractor = new DefaultMetadataExtractor();
+			extractor.metadataToExtract(MimeTypeUtils.TEXT_PLAIN, String.class, MetadataExtractor.ROUTE_KEY);
+			return extractor;
+		}
+
+		private DataBufferFactory initBufferFactory() {
+			return new NettyDataBufferFactory(PooledByteBufAllocator.DEFAULT);
 		}
 	}
 

spring-messaging/src/main/java/org/springframework/messaging/rsocket/RSocketRequester.java

@@ -28,8 +28,9 @@
 
 import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.core.ReactiveAdapterRegistry;
+import org.springframework.core.codec.Decoder;
 import org.springframework.lang.Nullable;
-import org.springframework.messaging.rsocket.annotation.support.AnnotationClientResponderConfigurer;
+import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
 import org.springframework.util.MimeType;
 
 /**
@@ -125,32 +126,32 @@ static RSocketRequester wrap(
 	interface Builder {
 
 		/**
-		 * Configure the MimeType for payload data which is then specified
-		 * on the {@code SETUP} frame and applies to the whole connection.
-		 * <p>By default this is set to the first concrete mime type supported
-		 * by the configured encoders and decoders.
-		 * @param mimeType the data MimeType to use
+		 * Configure the payload data MimeType to specify on the {@code SETUP}
+		 * frame that applies to the whole connection.
+		 * <p>If this is not set, the builder will try to select the mime type
+		 * based on the presence of a single
+		 * {@link RSocketStrategies.Builder#decoder(Decoder[])  non-default}
+		 * {@code Decoder}, or the first default decoder otherwise
+		 * (i.e. {@code String}) if no others are configured.
 		 */
 		RSocketRequester.Builder dataMimeType(@Nullable MimeType mimeType);
 
 		/**
-		 * Configure the MimeType for payload metadata which is then specified
-		 * on the {@code SETUP} frame and applies to the whole connection.
+		 * Configure the payload metadata MimeType to specify on the {@code SETUP}
+		 * frame and applies to the whole connection.
 		 * <p>By default this is set to
 		 * {@code "message/x.rsocket.composite-metadata.v0"} in which case the
 		 * route, if provided, is encoded as a
-		 * {@code "message/x.rsocket.routing.v0"} metadata entry, potentially
-		 * with other metadata entries added too. If this is set to any other
-		 * mime type, and a route is provided, it is assumed the mime type is
-		 * for the route.
-		 * @param mimeType the data MimeType to use
+		 * {@code "message/x.rsocket.routing.v0"} composite metadata entry.
+		 * For any other MimeType, it is assumed to be the MimeType for the
+		 * route, if provided.
 		 */
 		RSocketRequester.Builder metadataMimeType(MimeType mimeType);
 
 		/**
-		 * Set the {@link RSocketStrategies} to use for access to encoders,
-		 * decoders, and a factory for {@code DataBuffer's}.
-		 * @param strategies the codecs strategies to use
+		 * Set the {@link RSocketStrategies} to use.
+		 * <p>By default this is set to {@code RSocketStrategies.builder().build()}
+		 * but may be further customized via {@link #rsocketStrategies(Consumer)}.
 		 */
 		RSocketRequester.Builder rsocketStrategies(@Nullable RSocketStrategies strategies);
 
@@ -159,21 +160,20 @@ interface Builder {
 		 * <p>By default this starts out with an empty builder, i.e.
 		 * {@link RSocketStrategies#builder()}, but the strategies can also be
 		 * set via {@link #rsocketStrategies(RSocketStrategies)}.
-		 * @param configurer the configurer to apply
 		 */
 		RSocketRequester.Builder rsocketStrategies(Consumer<RSocketStrategies.Builder> configurer);
 
 		/**
 		 * Callback to configure the {@code ClientRSocketFactory} directly.
-		 * <p>See {@link AnnotationClientResponderConfigurer} for configuring a
-		 * client side responder.
+		 * <p>See static factory method
+		 * {@link RSocketMessageHandler#clientResponder(Object...)} for
+		 * configuring a client side responder with annotated methods.
 		 * <p><strong>Note:</strong> Do not set {@link #dataMimeType(MimeType)}
 		 * and {@link #metadataMimeType(MimeType)} directly on the
 		 * {@code ClientRSocketFactory}. Use the shortcuts on this builder
 		 * instead since the created {@code RSocketRequester} needs to be aware
 		 * of those settings.
-		 * @param configurer consumer to customize the factory
-		 * @see AnnotationClientResponderConfigurer
+		 * @see RSocketMessageHandler#clientResponder(Object...)
 		 */
 		RSocketRequester.Builder rsocketFactory(ClientRSocketFactoryConfigurer configurer);