Enhancements to content-type negotiation

- Added logic to wrap custom (user) message converters with NegotiatingMessageConverterWrapper
- Removed 'addDefaultConverters' flag from ContextFunctionCatalogAutoConfiguration as it is more confusing then useful
- Added test which uses wild card accept content-type with several converters available to ensure the appropriate one is used
- Made NegotiatingMessageConverterWrapper package private and moved it and it's test to a contex.config package

Resolves #462

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistry.java

@@ -720,7 +720,6 @@ public class BeanFactoryAwareFunctionRegistry
 						}
 					}
 				}
-
 			}
 
 			if (convertedValue == null) {
@@ -734,10 +733,10 @@ public class BeanFactoryAwareFunctionRegistry
 			Message outputMessage = null;
 			if (value instanceof Message) {
 				MessageHeaders headers = ((Message) value).getHeaders();
-				if (!headers.containsKey(NegotiatingMessageConverterWrapper.ACCEPT)) {
+				if (!headers.containsKey("accept")) {
 					Map<String, Object> headersMap = (Map<String, Object>) ReflectionUtils
 						.getField(this.headersField, headers);
-					headersMap.put(NegotiatingMessageConverterWrapper.ACCEPT, acceptedContentType);
+					headersMap.put("accept", acceptedContentType);
 					// Set the contentType header to the value of accept for "legacy" reasons. But, do not set the
 					// contentType header to the value of accept if it is a wildcard type, as this doesn't make sense.
 					// This also applies to the else branch below.
@@ -748,7 +747,7 @@ public class BeanFactoryAwareFunctionRegistry
 			}
 			else {
 				MessageBuilder<Object> builder = MessageBuilder.withPayload(value)
-					.setHeader(NegotiatingMessageConverterWrapper.ACCEPT, acceptedContentType);
+					.setHeader("accept", acceptedContentType);
 				if (acceptedContentType.isConcrete()) {
 					builder.setHeader(MessageHeaders.CONTENT_TYPE, acceptedContentType);
 				}

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java

@@ -38,7 +38,6 @@ import org.springframework.cloud.function.context.FunctionProperties;
 import org.springframework.cloud.function.context.FunctionRegistry;
 import org.springframework.cloud.function.context.catalog.BeanFactoryAwareFunctionRegistry;
 import org.springframework.cloud.function.context.catalog.FunctionInspector;
-import org.springframework.cloud.function.context.catalog.NegotiatingMessageConverterWrapper;
 import org.springframework.cloud.function.json.GsonMapper;
 import org.springframework.cloud.function.json.JacksonMapper;
 import org.springframework.context.ConfigurableApplicationContext;
@@ -51,6 +50,7 @@ import org.springframework.context.annotation.FilterType;
 import org.springframework.core.convert.converter.GenericConverter;
 import org.springframework.core.convert.support.ConfigurableConversionService;
 import org.springframework.lang.Nullable;
+import org.springframework.messaging.converter.AbstractMessageConverter;
 import org.springframework.messaging.converter.ByteArrayMessageConverter;
 import org.springframework.messaging.converter.CompositeMessageConverter;
 import org.springframework.messaging.converter.MappingJackson2MessageConverter;
@@ -84,13 +84,11 @@ public class ContextFunctionCatalogAutoConfiguration {
 
 		CompositeMessageConverter messageConverter = null;
 		List<MessageConverter> mcList = new ArrayList<>();
-		boolean addDefaultConverters = true;
 
 		if (!CollectionUtils.isEmpty(messageConverters)) {
 			for (MessageConverter mc : messageConverters) {
 				if (mc instanceof CompositeMessageConverter) {
 					mcList.addAll(((CompositeMessageConverter) mc).getConverters());
-					addDefaultConverters = false;
 				}
 				else {
 					mcList.add(mc);
@@ -99,17 +97,20 @@ public class ContextFunctionCatalogAutoConfiguration {
 		}
 
 		mcList = mcList.stream()
-				.filter(c -> isConverterEligible(c)).collect(Collectors.toList());
+				.filter(c -> isConverterEligible(c))
-		if (addDefaultConverters) {
+				.map(converter -> {
-			if (objectMapper == null) {
+					return converter instanceof AbstractMessageConverter
-				objectMapper = new ObjectMapper();
+							? NegotiatingMessageConverterWrapper.wrap((AbstractMessageConverter) converter)
-			}
+									: converter;
-			MappingJackson2MessageConverter jsonConverter = new MappingJackson2MessageConverter();
+				})
-			jsonConverter.setObjectMapper(objectMapper);
+				.collect(Collectors.toList());
-			mcList.add(NegotiatingMessageConverterWrapper.wrap(jsonConverter));
+
-			mcList.add(NegotiatingMessageConverterWrapper.wrap(new ByteArrayMessageConverter()));
+		MappingJackson2MessageConverter jsonConverter = new MappingJackson2MessageConverter();
-			mcList.add(NegotiatingMessageConverterWrapper.wrap(new StringMessageConverter()));
+		jsonConverter.setObjectMapper(objectMapper == null ? new ObjectMapper() : objectMapper);
-		}
+		mcList.add(NegotiatingMessageConverterWrapper.wrap(jsonConverter));
+		mcList.add(NegotiatingMessageConverterWrapper.wrap(new ByteArrayMessageConverter()));
+		mcList.add(NegotiatingMessageConverterWrapper.wrap(new StringMessageConverter()));
+
 		if (!CollectionUtils.isEmpty(mcList)) {
 			messageConverter = new CompositeMessageConverter(mcList);
 		}

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/NegotiatingMessageConverterWrapper.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.function.context.catalog;
+package org.springframework.cloud.function.context.config;
 
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
@@ -29,7 +29,7 @@ import org.springframework.util.MimeType;
  * contain a wildcard type (such as {@code text/*}, which may be tested against every
  * {@link AbstractMessageConverter#getSupportedMimeTypes() supported mime type} of the delegate MessageConverter.
  */
-public final class NegotiatingMessageConverterWrapper implements SmartMessageConverter {
+final class NegotiatingMessageConverterWrapper implements SmartMessageConverter {
 
 	/**
 	 * The Message Header key that may contain the list of (possibly wildcard) MimeTypes to convert to.

spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistryTests.java

@@ -20,6 +20,7 @@ package org.springframework.cloud.function.context.catalog;
 import java.io.Serializable;
 import java.lang.reflect.Field;
 import java.nio.charset.StandardCharsets;
+import java.util.Date;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Consumer;
@@ -41,13 +42,18 @@ import org.springframework.cloud.function.context.catalog.BeanFactoryAwareFuncti
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.converter.AbstractMessageConverter;
+import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.stereotype.Component;
+import org.springframework.util.MimeType;
 import org.springframework.util.ReflectionUtils;
 
+import static java.util.Collections.singletonList;
 import static org.assertj.core.api.Assertions.assertThat;
 
 /**
@@ -304,10 +310,10 @@ public class BeanFactoryAwareFunctionRegistryTests {
 	@Test
 	public void SCF_GH_409ConfigurationTests() {
 		FunctionCatalog catalog = this.configureCatalog(SCF_GH_409ConfigurationAsSupplier.class);
-		assertThat((Object) catalog.lookup("")).isNull();
+		assertThat((Function) catalog.lookup("")).isNull();
 
 		catalog = this.configureCatalog(SCF_GH_409ConfigurationAsFunction.class);
-		assertThat((Object) catalog.lookup("")).isNull();
+		assertThat((Function) catalog.lookup("")).isNull();
 	}
 
 	@Test
@@ -330,6 +336,121 @@ public class BeanFactoryAwareFunctionRegistryTests {
 		assertThat(composed).isFalse();
 	}
 
+
+	@SuppressWarnings("unchecked")
+	@Test
+	public void testSerializationWithCompatibleWildcardSubtypeAcceptHeader() {
+		FunctionCatalog catalog = this.configureCatalog(NegotiatingMessageConverterConfiguration.class);
+		FunctionInvocationWrapper function = catalog.lookup("echo", "text/*");
+
+		Message<Tuple2<String, String>> tupleResult = (Message<Tuple2<String, String>>) function.apply(MessageBuilder
+				.withPayload(Tuples.of("bonjour", "monde"))
+				.setHeader(MessageHeaders.CONTENT_TYPE, MimeType.valueOf("text/csv"))
+				.build()
+		);
+
+		assertThat(tupleResult.getHeaders().get(MessageHeaders.CONTENT_TYPE)).isEqualTo(MimeType.valueOf("text/csv"));
+		assertThat(tupleResult.getHeaders().get("accept")).isNull();
+
+		Message<Date> dateResult = (Message<Date>) function.apply(MessageBuilder
+				.withPayload(123)
+				.setHeader(MessageHeaders.CONTENT_TYPE, MimeType.valueOf("text/integer"))
+				.build()
+		);
+
+		assertThat(dateResult.getHeaders().get(MessageHeaders.CONTENT_TYPE)).isEqualTo(MimeType.valueOf("text/integer"));
+		assertThat(dateResult.getHeaders().get("accept")).isNull();
+	}
+
+	@EnableAutoConfiguration
+	public static class NegotiatingMessageConverterConfiguration {
+
+		@Bean
+		public Function<String, String> echo() {
+			return v -> v;
+		}
+
+		@Bean
+		public MessageConverter messageConverterA() {
+			return new ConverterA();
+		}
+
+		@Bean
+		public MessageConverter messageConverterB() {
+			return new ConverterB();
+		}
+
+
+		public static class ConverterB extends ConverterA {
+			ConverterB() {
+				super("text/integer");
+			}
+
+			@Override
+			protected Object convertFromInternal(
+					Message<?> message, Class<?> targetClass, @Nullable Object conversionHint) {
+				return message.getPayload().toString();
+			}
+
+			@Override
+			public Object convertToInternal(Object rawPayload, MessageHeaders headers, Object conversionHint) {
+				return Integer.parseInt((String) rawPayload);
+			}
+
+			@Override
+			protected boolean canConvertFrom(Message<?> message, @Nullable Class<?> targetClass) {
+				return supportsMimeType(message.getHeaders()) && String.class.isAssignableFrom(targetClass)
+						&& message.getPayload() instanceof Integer;
+			}
+
+			@Override
+			protected boolean canConvertTo(Object payload, @Nullable MessageHeaders headers) {
+				return payload instanceof String;
+			}
+		}
+
+		private static class ConverterA extends AbstractMessageConverter {
+
+			ConverterA() {
+				this("text/csv");
+			}
+
+			ConverterA(String mimeType) {
+				super(singletonList(MimeType.valueOf(mimeType)));
+			}
+
+			@Override
+			protected Object convertFromInternal(
+					Message<?> message, Class<?> targetClass, @Nullable Object conversionHint) {
+				Tuple2<String, String> payload = (Tuple2<String, String>) message.getPayload();
+
+				String convertedPayload = payload.getT1() + "," + payload.getT2();
+				return convertedPayload;
+			}
+
+			@Override
+			public Object convertToInternal(Object rawPayload, MessageHeaders headers, Object conversionHint) {
+				return Tuples.fromArray(((String) rawPayload).split(","));
+			}
+
+			@Override
+			protected boolean canConvertFrom(Message<?> message, @Nullable Class<?> targetClass) {
+				return supportsMimeType(message.getHeaders()) && String.class.isAssignableFrom(targetClass)
+						&& message.getPayload() instanceof Tuple2;
+			}
+
+			@Override
+			protected boolean canConvertTo(Object payload, @Nullable MessageHeaders headers) {
+				return payload instanceof String && ((String) payload).split(",").length == 2;
+			}
+
+			@Override
+			protected boolean supports(Class<?> clazz) {
+				throw new UnsupportedOperationException();
+			}
+		}
+	}
+
 	@EnableAutoConfiguration
 	@Configuration
 	protected static class SampleFunctionConfiguration {
@@ -526,9 +647,6 @@ public class BeanFactoryAwareFunctionRegistryTests {
 				// TODO Auto-generated method stub
 				return null;
 			}
-
-
-
 		}
 	}
 

spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/config/NegotiatingMessageConverterWrapperTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2020 the original author or authors.
+ * Copyright 2020-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.
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.function.context.catalog;
+package org.springframework.cloud.function.context.config;
 
 import java.util.Collection;
 import java.util.Collections;
@@ -39,10 +39,15 @@ import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.util.Maps.newHashMap;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
-import static org.springframework.cloud.function.context.catalog.NaiveCsvTupleMessageConverter.MAGIC_NULL;
+import static org.springframework.cloud.function.context.config.NaiveCsvTupleMessageConverter.MAGIC_NULL;
-import static org.springframework.cloud.function.context.catalog.NegotiatingMessageConverterWrapper.ACCEPT;
+import static org.springframework.cloud.function.context.config.NegotiatingMessageConverterWrapper.ACCEPT;
 import static org.springframework.messaging.MessageHeaders.CONTENT_TYPE;
 
+/**
+ *
+ * @author Florent Biville
+ *
+ */
 public class NegotiatingMessageConverterWrapperTests {
 
 	Collection<Tuple2<?, ?>> somePayload = asList(Tuples.of("hello", "world"), Tuples.of("bonjour", "monde"));