Support hints for codec on client and server side

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

@@ -20,7 +20,6 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
-import java.util.Map;
 import java.util.function.Consumer;
 
 import org.apache.commons.logging.Log;
@@ -226,13 +225,11 @@ private Mono<Object> decodeContent(MethodParameter parameter, Message<?> message
 		isContentRequired = isContentRequired || (adapter != null && !adapter.supportsEmpty());
 		Consumer<Object> validator = getValidator(message, parameter);
 
-		Map<String, Object> hints = Collections.emptyMap();
-
 		for (Decoder<?> decoder : this.decoders) {
 			if (decoder.canDecode(elementType, mimeType)) {
 				if (adapter != null && adapter.isMultiValue()) {
 					Flux<?> flux = content
-							.map(buffer -> decoder.decode(buffer, elementType, mimeType, hints))
+							.map(buffer -> decoder.decode(buffer, elementType, mimeType, message.getHeaders()))
 							.onErrorResume(ex -> Flux.error(handleReadError(parameter, message, ex)));
 					if (isContentRequired) {
 						flux = flux.switchIfEmpty(Flux.error(() -> handleMissingBody(parameter, message)));
@@ -245,7 +242,7 @@ private Mono<Object> decodeContent(MethodParameter parameter, Message<?> message
 				else {
 					// Single-value (with or without reactive type wrapper)
 					Mono<?> mono = content.next()
-							.map(buffer -> decoder.decode(buffer, elementType, mimeType, hints))
+							.map(buffer -> decoder.decode(buffer, elementType, mimeType,  message.getHeaders()))
 							.onErrorResume(ex -> Mono.error(handleReadError(parameter, message, ex)));
 					if (isContentRequired) {
 						mono = mono.switchIfEmpty(Mono.error(() -> handleMissingBody(parameter, message)));

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

@@ -0,0 +1,179 @@
+/*
+ * Copyright 2002-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.messaging.rsocket;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.CompositeByteBuf;
+import io.rsocket.metadata.CompositeMetadata;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.codec.Decoder;
+import org.springframework.core.codec.Encoder;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.core.io.buffer.NettyDataBufferFactory;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+import org.springframework.util.MimeType;
+import org.springframework.util.MimeTypeUtils;
+
+import static org.springframework.messaging.rsocket.DefaultRSocketRequester.CODEC_HINTS_MIME;
+
+/**
+ *
+ * Default implement for {@link HintMetadataCodec}.
+ *
+ * @author Rudy Steiner
+ * @since 5.3.2
+ **/
+public class DefaultHintMetadataCodec implements HintMetadataCodec {
+	private static final Log logger = LogFactory.getLog(DefaultHintMetadataCodec.class);
+	private static final Map<String, Object> EMPTY_HINTS = Collections.emptyMap();
+	private final NettyDataBufferFactory nettyDataBufferFactory ;
+	private final DataBufferFactory bufferFactory;
+	private ByteBufAllocator byteBufAllocator;
+	private List<Encoder<?>> encoders;
+	private List<Decoder<?>> decoders;
+
+	public DefaultHintMetadataCodec(DataBufferFactory dataBufferFactory, MetadataExtractor metadataExtractor,
+									List<Encoder<?>> encoders, List<Decoder<?>> decoders){
+		Assert.notNull(dataBufferFactory, "'dataBufferFactory' must not be null");
+		Assert.notNull(encoders, "'encoders' must not be null");
+		Assert.notNull(decoders, "'decoders' must not be null");
+		this.bufferFactory = dataBufferFactory;
+		this.encoders = encoders;
+		this.decoders = decoders;
+		this.byteBufAllocator = allocator();
+		this.nettyDataBufferFactory = new NettyDataBufferFactory(this.byteBufAllocator);
+		if (metadataExtractor instanceof MetadataExtractorRegistry) {
+			try {
+				MetadataExtractorRegistry registry = (MetadataExtractorRegistry) metadataExtractor;
+				registry.metadataToExtract(CODEC_HINTS_MIME, ByteBuf.class, this::decodeHints);
+			}
+			catch (Throwable ex){
+				logger.error("Register hints extractor failed", ex);
+			}
+		}
+		else if(logger.isDebugEnabled()){
+			logger.debug("Failed to register extractor for "+ CODEC_HINTS_MIME.toString());
+		}
+	}
+	@Override
+	public void encodeHints(Map<String, Object> hints, Consumer<ByteBuf> consumer) {
+		CompositeByteBuf composite = this.byteBufAllocator.compositeBuffer();
+		try {
+			hints.forEach((key, value) -> {
+				Assert.notNull(key, "'key' must not be null");
+				Assert.notNull(value, "'value' must not be null");
+				encodeHint(composite, key, value);
+			});
+			consumer.accept(composite);
+		}
+		catch (Throwable ex){
+			composite.release();
+			throw ex;
+		}
+	}
+
+
+
+	@Override
+	public void decodeHints(ByteBuf hints, Map<String, Object> result) {
+		for (CompositeMetadata.Entry entry : new CompositeMetadata(hints, false)) {
+			try {
+				consumeHint(entry.getMimeType(),(key, resolvableType)->{
+					Assert.isTrue(result.get(resolvableType) == null,"Duplicate key:" + key);
+					decodeHint(key,resolvableType,entry.getContent(),result);
+				});
+			}
+			catch (Throwable ex){
+				logger.error("Extract hints exception", ex);
+			}
+		}
+	}
+
+	private void encodeHint(CompositeByteBuf composite,String key,Object value){
+		io.rsocket.metadata.CompositeMetadataCodec.encodeAndAddMetadata(composite, this.byteBufAllocator,
+				keyValueClass(key, value),
+				PayloadUtils.asByteBuf(encodeData(value, ResolvableType.forInstance(value),null, MimeTypeUtils.ALL)));
+	}
+
+	private void decodeHint(String key, ResolvableType resolvableType , ByteBuf value, Map<String,Object> result){
+		Decoder<Object> decoder = this.decoder( resolvableType, MimeTypeUtils.ALL);
+		result.put(key, decoder.decode(this.nettyDataBufferFactory.wrap(value.retain()), resolvableType,
+				MimeTypeUtils.ALL, null));
+	}
+
+	@SuppressWarnings("unchecked")
+	private <T> DataBuffer encodeData(T value, ResolvableType elementType, @Nullable Encoder<?> encoder, MimeType dataMimeType) {
+		if (encoder == null) {
+			elementType = ResolvableType.forInstance(value);
+			encoder = this.encoder(elementType, dataMimeType);
+		}
+		return ((Encoder<T>) encoder).encodeValue(
+				value, this.bufferFactory, elementType, dataMimeType,EMPTY_HINTS);
+	}
+	private ByteBufAllocator allocator() {
+		return this.bufferFactory instanceof NettyDataBufferFactory ?
+				((NettyDataBufferFactory) this.bufferFactory).getByteBufAllocator() : ByteBufAllocator.DEFAULT;
+	}
+	/**
+	 * Composite hint key and class name of value object.
+	 **/
+	private String keyValueClass(String key, Object object){
+		return key +"/" +object.getClass().getName();
+	}
+
+	/**
+	 * Consume hint keyValueClass.
+	 **/
+	private void consumeHint(String keyValueClass, BiConsumer<String, ResolvableType> keyValueResolveType) throws ClassNotFoundException{
+		String[]  keyClass = keyValueClass.split("/");
+		Assert.isTrue(keyClass.length ==2 ,"Invalid key value class name:"+keyValueClass);
+		keyValueResolveType.accept(keyClass[0],ResolvableType.forClass(Class.forName(keyClass[1])));
+	}
+
+	@SuppressWarnings("unchecked")
+	private  <T> Decoder<T> decoder(ResolvableType elementType, @Nullable MimeType mimeType) {
+		for (Decoder<?> decoder : this.decoders) {
+			if (decoder.canDecode(elementType, mimeType)) {
+				return (Decoder<T>) decoder;
+			}
+		}
+		throw new IllegalArgumentException("No decoder for " + elementType);
+	}
+
+	@SuppressWarnings("unchecked")
+	private  <T> Encoder<T> encoder(ResolvableType elementType, @Nullable MimeType mimeType) {
+		for (Encoder<?> encoder : this.encoders) {
+			if (encoder.canEncode(elementType, mimeType)) {
+				return (Encoder<T>) encoder;
+			}
+		}
+		throw new IllegalArgumentException("No encoder for " + elementType);
+	}
+
+}

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

@@ -18,11 +18,14 @@
 
 import java.util.Collections;
 import java.util.Map;
+import java.util.Objects;
 import java.util.function.Consumer;
 
 import io.rsocket.Payload;
 import io.rsocket.RSocket;
 import io.rsocket.core.RSocketClient;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -46,8 +49,9 @@
  * @since 5.2
  */
 final class DefaultRSocketRequester implements RSocketRequester {
-
+	private static final Log logger = LogFactory.getLog(DefaultRSocketRequester.class);
 	private static final Map<String, Object> EMPTY_HINTS = Collections.emptyMap();
+	public static final MimeType CODEC_HINTS_MIME = MimeType.valueOf("message/x.codec.hints.v0");
 
 	private final RSocketClient rsocketClient;
 
@@ -131,6 +135,9 @@ private class DefaultRequestSpec implements RequestSpec {
 		@Nullable
 		private Flux<Payload> payloadFlux;
 
+		@Nullable
+		private Map<String, Object> hints;
+
 
 		public DefaultRequestSpec(String route, Object... vars) {
 			this.metadataEncoder.route(route, vars);
@@ -153,6 +160,25 @@ public RequestSpec metadata(Consumer<MetadataSpec<?>> configurer) {
 			return this;
 		}
 
+		@Override
+		public RequestSpec hints(Map<String, Object> hints) {
+			Assert.isNull(this.hints, "'hints' only can be set once");
+			Assert.notNull(hints, "'hints' must not be null");
+			Assert.isTrue(hints.size() > 0 , "'hints' must not be empty");
+			if (logger.isDebugEnabled()){
+				logger.debug("Attach codec hint keys:" + hints.keySet() + ".");
+			}
+			this.hints = hints;
+			if (Objects.isNull(strategies.hintMetadataCodec())) {
+				if (logger.isDebugEnabled()) {
+					logger.debug("No hints metadata codec, hints only available on client side");
+				}
+				return this;
+			}
+			strategies.hintMetadataCodec().encodeHints(hints, e -> this.metadata(e, CODEC_HINTS_MIME));
+			return this;
+		}
+
 		@Override
 		public RequestSpec data(Object data) {
 			Assert.notNull(data, "'data' must not be null");
@@ -170,6 +196,7 @@ public RequestSpec data(Object producer, Class<?> elementClass) {
 			return this;
 		}
 
+
 		@Nullable
 		private ReactiveAdapter getAdapter(Class<?> aClass) {
 			return strategies.reactiveAdapterRegistry().getAdapter(aClass);
@@ -196,7 +223,7 @@ else if (adapter != null) {
 			}
 			else {
 				ResolvableType type = ResolvableType.forInstance(input);
-				this.payloadMono = firstPayload(Mono.fromCallable(() -> encodeData(input, type, null)));
+				this.payloadMono = firstPayload(Mono.fromCallable(() -> encodeData(input, type, null, this.hints)));
 				this.payloadFlux = null;
 				return;
 			}
@@ -212,7 +239,7 @@ else if (adapter != null) {
 
 			if (adapter != null && !adapter.isMultiValue()) {
 				Mono<DataBuffer> data = Mono.from(publisher)
-						.map(value -> encodeData(value, elementType, encoder))
+						.map(value -> encodeData(value, elementType, encoder, this.hints))
 						.switchIfEmpty(emptyBufferMono);
 				this.payloadMono = firstPayload(data);
 				this.payloadFlux = null;
@@ -221,7 +248,7 @@ else if (adapter != null) {
 
 			this.payloadMono = null;
 			this.payloadFlux = Flux.from(publisher)
-					.map(value -> encodeData(value, elementType, encoder))
+					.map(value -> encodeData(value, elementType, encoder, this.hints))
 					.switchIfEmpty(emptyBufferMono)
 					.switchOnFirst((signal, inner) -> {
 						DataBuffer data = signal.get();
@@ -237,13 +264,26 @@ else if (adapter != null) {
 		}
 
 		@SuppressWarnings("unchecked")
-		private <T> DataBuffer encodeData(T value, ResolvableType elementType, @Nullable Encoder<?> encoder) {
+		private <T> DataBuffer encodeData(T value, ResolvableType elementType, @Nullable Encoder<?> encoder,
+										Map<String, Object> hints) {
+			return encodeData(value,elementType,encoder,dataMimeType , hints);
+		}
+
+		@SuppressWarnings("unchecked")
+		private <T> DataBuffer encodeData(T value, ResolvableType elementType, @Nullable Encoder<?> encoder,
+		MimeType dataMimeType) {
+			return encodeData(value,elementType,encoder,dataMimeType, EMPTY_HINTS);
+		}
+
+		@SuppressWarnings("unchecked")
+		private <T> DataBuffer encodeData(T value, ResolvableType elementType, @Nullable Encoder<?> encoder,
+										MimeType dataMimeType, Map<String,Object> hints) {
 			if (encoder == null) {
 				elementType = ResolvableType.forInstance(value);
 				encoder = strategies.encoder(elementType, dataMimeType);
 			}
 			return ((Encoder<T>) encoder).encodeValue(
-					value, bufferFactory(), elementType, dataMimeType, EMPTY_HINTS);
+					value, bufferFactory(), elementType, dataMimeType, hints);
 		}
 
 		/**