GH-1594 Ensured Function conversion honors outbound CT

Ensured that Message conversion that happenes within FunctionInvoker honors the outbound binding content-type in the event MessageConverter itself did not sent one.
Restructured some code/API around function invocation
Added tests

Resolves #1594

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java

@@ -16,7 +16,6 @@
 
 package org.springframework.cloud.stream.binder;
 
-import java.lang.reflect.Method;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.function.Consumer;
@@ -65,7 +64,6 @@ import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.support.ChannelInterceptor;
 import org.springframework.retry.RecoveryCallback;
 import org.springframework.util.Assert;
-import org.springframework.util.ReflectionUtils;
 import org.springframework.util.StringUtils;
 
 /**
@@ -854,7 +852,6 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 				IntegrationFlowBuilder integrationFlowBuilder = IntegrationFlows.from(outputChannel).bridge();
 				publisher = integrationFlowBuilder.toReactivePublisher();
 			}
-			this.propagateProducerPropertiesToFunction(producerProperties);
 			if (this.integrationFlowFunctionSupport.containsFunction(Function.class,
 					this.streamFunctionProperties.getDefinition())) {
 				DirectChannel actualOutputChannel = new DirectChannel();
@@ -877,7 +874,6 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 			if (inputChannel instanceof AbstractMessageChannel) {
 				moveChannelInterceptors((AbstractMessageChannel) inputChannel, actualInputChannel);
 			}
-			this.propagateConsumerPropertiesToFunction(consumerProperties);
 			this.integrationFlowFunctionSupport.andThenFunction(MessageChannelReactiveUtils.toPublisher(actualInputChannel),
 					inputChannel, this.streamFunctionProperties);
 			return actualInputChannel;
@@ -885,32 +881,6 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		return (SubscribableChannel) inputChannel;
 	}
 
-	// we're doing it reflectively so we don't expose this as a property to the user
-	private void propagateProducerPropertiesToFunction(ProducerProperties producerProperties) {
-		try {
-			Method setProducerProperties = ReflectionUtils.findMethod(StreamFunctionProperties.class,
-					"setProducerProperties", ProducerProperties.class);
-			setProducerProperties.setAccessible(true);
-			setProducerProperties.invoke(this.streamFunctionProperties, producerProperties);
-		}
-		catch (Exception e) {
-			throw new IllegalStateException(e);
-		}
-	}
-
-	// we're doing it reflectively so we don't expose this as a property to the user
-	private void propagateConsumerPropertiesToFunction(ConsumerProperties consumerProperties) {
-		try {
-			Method setConsumerProperties = ReflectionUtils.findMethod(StreamFunctionProperties.class,
-					"setConsumerProperties", ConsumerProperties.class);
-			setConsumerProperties.setAccessible(true);
-			setConsumerProperties.invoke(this.streamFunctionProperties, consumerProperties);
-		}
-		catch (Exception e) {
-			throw new IllegalStateException(e);
-		}
-	}
-
 	private void moveChannelInterceptors(AbstractMessageChannel existingMessageChannel,
 			AbstractMessageChannel actualMessageChannel) {
 		for (ChannelInterceptor channelInterceptor : existingMessageChannel.getChannelInterceptors()) {

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java

@@ -301,9 +301,17 @@ public class MessageConverterConfigurer implements MessageChannelAndSourceConfig
 			this.messageConverter = messageConverter;
 		}
 
-
 		@Override
 		public Message<?> doPreSend(Message<?> message, MessageChannel channel) {
+
+			// If handler is a function, FunctionInvoker will already perform message conversion.
+			// In fact in the future we should consider propagating knowledge of the default content type
+			//to MessageConverters instead of interceptors
+			if (message.getPayload() instanceof byte[] && message.getHeaders().containsKey(MessageHeaders.CONTENT_TYPE)) {
+				return message;
+			}
+
+
 			// ===== 1.3 backward compatibility code part-1 ===
 			String oct = message.getHeaders().containsKey(MessageHeaders.CONTENT_TYPE) ? message.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString() : null;
 			String ct = message.getPayload() instanceof String

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -16,7 +16,9 @@
 
 package org.springframework.cloud.stream.function;
 
+import java.lang.reflect.Field;
 import java.time.Duration;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -30,6 +32,7 @@ import org.springframework.cloud.function.context.FunctionType;
 import org.springframework.cloud.function.context.catalog.FunctionInspector;
 import org.springframework.cloud.stream.binder.ConsumerProperties;
 import org.springframework.cloud.stream.binder.ProducerProperties;
+import org.springframework.cloud.stream.config.BindingServiceProperties;
 import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
@@ -38,6 +41,8 @@ import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.converter.CompositeMessageConverter;
 import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.util.Assert;
+import org.springframework.util.MimeType;
+import org.springframework.util.ReflectionUtils;
 
 /**
  *
@@ -54,6 +59,13 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
 
 	private static final Log logger = LogFactory.getLog(FunctionInvoker.class);
 
+	private static final Field MESSAGE_HEADERS_FIELD;
+
+	static {
+		MESSAGE_HEADERS_FIELD = ReflectionUtils.findField(MessageHeaders.class, "headers");
+		MESSAGE_HEADERS_FIELD.setAccessible(true);
+	}
+
 	private final Class<?> inputClass;
 
 	private final Class<?> outputClass;
@@ -70,6 +82,10 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
 
 	private final ProducerProperties producerProperties;
 
+	private final BindingServiceProperties bindingServiceProperties;
+
+	private final StreamFunctionProperties functionProperties;
+
 	FunctionInvoker(StreamFunctionProperties functionProperties, FunctionCatalogWrapper functionCatalog, FunctionInspector functionInspector,
 			CompositeMessageConverterFactory compositeMessageConverterFactory) {
 		this(functionProperties, functionCatalog, functionInspector, compositeMessageConverterFactory, null);
@@ -78,6 +94,8 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	FunctionInvoker(StreamFunctionProperties functionProperties, FunctionCatalogWrapper functionCatalog, FunctionInspector functionInspector,
 			CompositeMessageConverterFactory compositeMessageConverterFactory, MessageChannel errorChannel) {
+
+		this.functionProperties = functionProperties;
 		Object originalUserFunction = functionCatalog.lookup(functionProperties.getDefinition());
 
 		this.userFunction = originalUserFunction instanceof Consumer
@@ -91,10 +109,9 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
 		this.inputClass = functionType.getInputType();
 		this.outputClass = functionType.getOutputType();
 		this.errorChannel = errorChannel;
-		this.consumerProperties = functionProperties.getConsumerProperties() == null
-				? new ConsumerProperties() : functionProperties.getConsumerProperties();
-		this.producerProperties = functionProperties.getProducerProperties() == null
-				? new ProducerProperties() : functionProperties.getProducerProperties();
+		this.bindingServiceProperties = functionProperties.getBindingServiceProperties();
+		this.consumerProperties = bindingServiceProperties.getConsumerProperties(functionProperties.getInputDestinationName());
+		this.producerProperties = bindingServiceProperties.getProducerProperties(functionProperties.getOutputDestinationName());
 	}
 
 	@Override
@@ -145,16 +162,22 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
 					(value instanceof Message
 							? value
 							: this.messageConverter.toMessage(value, originalMessage.getHeaders(), this.outputClass));
-			if (returnMessage == null) {
-				if (value.getClass().isAssignableFrom(this.outputClass)) {
-					returnMessage = wrapOutputToMessage(value, originalMessage);
-				}
+			if (returnMessage == null && value.getClass().isAssignableFrom(this.outputClass)) {
+				returnMessage = wrapOutputToMessage(value, originalMessage);
+			}
+			else if (this.bindingServiceProperties != null
+						&& this.bindingServiceProperties.getBindingProperties(this.functionProperties.getOutputDestinationName()) != null
+						&& !returnMessage.getHeaders().containsKey(MessageHeaders.CONTENT_TYPE)) {
+
+				((Map<String, Object>) ReflectionUtils.getField(MESSAGE_HEADERS_FIELD, returnMessage.getHeaders()))
+						.put(MessageHeaders.CONTENT_TYPE, MimeType.valueOf(bindingServiceProperties.getBindingProperties("output").getContentType()));
 			}
 			Assert.notNull(returnMessage, "Failed to convert result value '" + value + "' to message.");
 		}
 		return returnMessage;
 	}
 
+
 	@SuppressWarnings("unchecked")
 	private <T> Message<O> wrapOutputToMessage(T value, Message<I> originalMessage) {
 		Message<O> returnMessage = (Message<O>) MessageBuilder.withPayload(value).copyHeaders(originalMessage.getHeaders())

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -58,8 +58,6 @@ public class IntegrationFlowFunctionSupport {
 
 	private final StreamFunctionProperties functionProperties;
 
-	private final BindingServiceProperties bindingServiceProperties;
-
 	@Autowired
 	private MessageChannel errorChannel;
 
@@ -81,7 +79,7 @@ public class IntegrationFlowFunctionSupport {
 		this.functionInspector = functionInspector;
 		this.messageConverterFactory = messageConverterFactory;
 		this.functionProperties = functionProperties;
-		this.bindingServiceProperties = bindingServiceProperties;
+		this.functionProperties.setBindingServiceProperties(bindingServiceProperties);
 	}
 
 	/**
@@ -158,14 +156,14 @@ public class IntegrationFlowFunctionSupport {
 		if (inputChannel instanceof IntegrationObjectSupport) {
 			String inputBindingName = ((IntegrationObjectSupport)inputChannel).getComponentName();
 			if (StringUtils.hasText(inputBindingName)) {
-				this.functionProperties.setConsumerProperties(this.bindingServiceProperties.getConsumerProperties(inputBindingName));
+				this.functionProperties.setInputDestinationName(inputBindingName);
 			}
 		}
 
 		if (outputChannel instanceof IntegrationObjectSupport) {
 			String outputBindingName = ((IntegrationObjectSupport)outputChannel).getComponentName();
 			if (StringUtils.hasText(outputBindingName)) {
-				this.functionProperties.setProducerProperties(this.bindingServiceProperties.getProducerProperties(outputBindingName));
+				this.functionProperties.setOutputDestinationName(outputBindingName);
 			}
 		}
 

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -17,8 +17,8 @@
 package org.springframework.cloud.stream.function;
 
 import org.springframework.boot.context.properties.ConfigurationProperties;
-import org.springframework.cloud.stream.binder.ConsumerProperties;
-import org.springframework.cloud.stream.binder.ProducerProperties;
+import org.springframework.cloud.stream.config.BindingServiceProperties;
+import org.springframework.cloud.stream.messaging.Processor;
 
 /**
  *
@@ -36,9 +36,11 @@ public class StreamFunctionProperties {
 	 */
 	private String definition;
 
-	private ConsumerProperties consumerProperties;
+	private BindingServiceProperties bindingServiceProperties;
 
-	private ProducerProperties producerProperties;
+	private String inputDestinationName = Processor.INPUT;
+
+	private String outputDestinationName = Processor.OUTPUT;
 
 	public String getDefinition() {
 		return this.definition;
@@ -48,19 +50,27 @@ public class StreamFunctionProperties {
 		this.definition = definition;
 	}
 
-	protected ConsumerProperties getConsumerProperties() {
-		return consumerProperties;
+	BindingServiceProperties getBindingServiceProperties() {
+		return bindingServiceProperties;
 	}
 
-	void setConsumerProperties(ConsumerProperties consumerProperties) {
-		this.consumerProperties = consumerProperties;
+	void setBindingServiceProperties(BindingServiceProperties bindingServiceProperties) {
+		this.bindingServiceProperties = bindingServiceProperties;
 	}
 
-	protected ProducerProperties getProducerProperties() {
-		return producerProperties;
+	String getInputDestinationName() {
+		return inputDestinationName;
 	}
 
-	void setProducerProperties(ProducerProperties producerProperties) {
-		this.producerProperties = producerProperties;
+	void setInputDestinationName(String inputDestinationName) {
+		this.inputDestinationName = inputDestinationName;
+	}
+
+	String getOutputDestinationName() {
+		return outputDestinationName;
+	}
+
+	void setOutputDestinationName(String outputDestinationName) {
+		this.outputDestinationName = outputDestinationName;
 	}
 }

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -30,16 +30,22 @@ import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.cloud.function.context.FunctionCatalog;
 import org.springframework.cloud.function.context.catalog.FunctionInspector;
-import org.springframework.cloud.stream.binder.ConsumerProperties;
-import org.springframework.cloud.stream.binder.ProducerProperties;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.StreamMessageConverter;
+import org.springframework.cloud.stream.binder.test.InputDestination;
+import org.springframework.cloud.stream.binder.test.OutputDestination;
 import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
+import org.springframework.cloud.stream.config.BindingServiceProperties;
 import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
 import org.springframework.cloud.stream.function.pojo.Baz;
 import org.springframework.cloud.stream.function.pojo.ErrorBaz;
+import org.springframework.cloud.stream.messaging.Processor;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.util.ReflectionUtils;
 
@@ -54,6 +60,112 @@ import static org.junit.Assert.assertEquals;
  */
 public class FunctionInvokerTests {
 
+	@Test
+	public void testFunctionHonorsOutboundBindingContentType() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(ConverterDoesNotProduceCTConfiguration.class)).web(
+				WebApplicationType.NONE)
+				.run("--spring.jmx.enabled=false",
+						"--spring.cloud.stream.function.definition=func",
+						"--spring.cloud.stream.bindings.output.contentType=text/plain")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context.getBean(OutputDestination.class);
+
+			Message<byte[]> inputMessage = MessageBuilder.withPayload("{\"name\":\"bob\"}".getBytes())
+					.setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar").build();
+			inputDestination.send(inputMessage);
+
+			Message<byte[]> outputMessage = outputDestination.receive();
+			assertEquals("text/plain", outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString());
+
+		}
+	}
+
+	@EnableAutoConfiguration
+	@EnableBinding(Processor.class)
+	public static class ConverterDoesNotProduceCTConfiguration {
+
+		@Bean
+		public Function<String, String> func() {
+			return x -> x;
+		}
+
+		@StreamMessageConverter
+		@Bean
+		public MessageConverter customConverter() {
+			return new MessageConverter() {
+
+				@Override
+				public Message<?> toMessage(Object payload, MessageHeaders headers) {
+					return new GenericMessage<byte[]>(((String)payload).getBytes());
+				}
+
+				@Override
+				public Object fromMessage(Message<?> message, Class<?> targetClass) {
+					String contentType = (String) message.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString();
+					if (contentType.equals("foo/bar")) {
+						return new String((byte[])message.getPayload());
+					}
+					return null;
+				}
+			};
+		}
+	}
+
+	@Test
+	public void testFunctionHonorsConverterSetContentType() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(ConverterInjectingCTConfiguration.class)).web(
+				WebApplicationType.NONE)
+				.run("--spring.jmx.enabled=false",
+						"--spring.cloud.stream.function.definition=func",
+						"--spring.cloud.stream.bindings.output.contentType=text/plain")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context.getBean(OutputDestination.class);
+
+			Message<byte[]> inputMessage = MessageBuilder.withPayload("{\"name\":\"bob\"}".getBytes())
+					.setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar").build();
+			inputDestination.send(inputMessage);
+
+			Message<byte[]> outputMessage = outputDestination.receive();
+			assertEquals("ping/pong", outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString());
+
+		}
+	}
+
+	@EnableAutoConfiguration
+	@EnableBinding(Processor.class)
+	public static class ConverterInjectingCTConfiguration {
+
+		@Bean
+		public Function<String, String> func() {
+			return x -> x;
+		}
+
+		@StreamMessageConverter
+		@Bean
+		public MessageConverter customConverter() {
+			return new MessageConverter() {
+
+				@Override
+				public Message<?> toMessage(Object payload, MessageHeaders headers) {
+					return MessageBuilder.withPayload(((String)payload).getBytes()).setHeader(MessageHeaders.CONTENT_TYPE, "ping/pong").build();
+				}
+
+				@Override
+				public Object fromMessage(Message<?> message, Class<?> targetClass) {
+					String contentType = (String) message.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString();
+					if (contentType.equals("foo/bar")) {
+						return new String((byte[])message.getPayload());
+					}
+					return null;
+				}
+			};
+		}
+	}
+
 	@Test
 	public void testSameMessageTypesAreNotConverted() {
 		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
@@ -176,12 +288,14 @@ public class FunctionInvokerTests {
 
 	private StreamFunctionProperties createStreamFunctionProperties() {
 		StreamFunctionProperties functionProperties = new StreamFunctionProperties();
-		ConsumerProperties consumerProperties = new ConsumerProperties();
-		consumerProperties.setMaxAttempts(3);
+		functionProperties.setInputDestinationName("input");
+		functionProperties.setOutputDestinationName("output");
+		BindingServiceProperties bindingServiceProperties = new BindingServiceProperties();
+		bindingServiceProperties.getConsumerProperties("input").setMaxAttempts(3);
 		try {
-			Field f = ReflectionUtils.findField(StreamFunctionProperties.class, "consumerProperties");
+			Field f = ReflectionUtils.findField(StreamFunctionProperties.class, "bindingServiceProperties");
 			f.setAccessible(true);
-			f.set(functionProperties, consumerProperties);
+			f.set(functionProperties, bindingServiceProperties);
 			return functionProperties;
 		} catch (Exception e) {
 			throw new IllegalStateException(e);
@@ -190,17 +304,15 @@ public class FunctionInvokerTests {
 
 	private StreamFunctionProperties createStreamFunctionPropertiesWithNativeEncoding() {
 		StreamFunctionProperties functionProperties = new StreamFunctionProperties();
-		ConsumerProperties consumerProperties = new ConsumerProperties();
-		consumerProperties.setMaxAttempts(3);
-		ProducerProperties producerProperties = new ProducerProperties();
-		producerProperties.setUseNativeEncoding(true);
+		functionProperties.setInputDestinationName("input");
+		functionProperties.setOutputDestinationName("output");
+		BindingServiceProperties bindingServiceProperties = new BindingServiceProperties();
+		bindingServiceProperties.getConsumerProperties("input").setMaxAttempts(3);
+		bindingServiceProperties.getProducerProperties("output").setUseNativeEncoding(true);
 		try {
-			Field c = ReflectionUtils.findField(StreamFunctionProperties.class, "consumerProperties");
-			Field p = ReflectionUtils.findField(StreamFunctionProperties.class, "producerProperties");
-			c.setAccessible(true);
-			c.set(functionProperties, consumerProperties);
-			p.setAccessible(true);
-			p.set(functionProperties, producerProperties);
+			Field bspField = ReflectionUtils.findField(StreamFunctionProperties.class, "bindingServiceProperties");
+			bspField.setAccessible(true);
+			bspField.set(functionProperties, bindingServiceProperties);
 			return functionProperties;
 		} catch (Exception e) {
 			throw new IllegalStateException(e);