Fixed native encoding for output of invoked function.

Resolves #1536

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

@@ -22,13 +22,13 @@ import java.util.function.Function;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 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.converter.CompositeMessageConverterFactory;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
@@ -66,6 +66,8 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
 
 	private final ConsumerProperties consumerProperties;
 
+	private final ProducerProperties producerProperties;
+
 	FunctionInvoker(StreamFunctionProperties functionProperties, FunctionCatalogWrapper functionCatalog, FunctionInspector functionInspector,
 			CompositeMessageConverterFactory compositeMessageConverterFactory) {
 		this(functionProperties, functionCatalog, functionInspector, compositeMessageConverterFactory, null);
@@ -83,6 +85,8 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
 		this.errorChannel = errorChannel;
 		this.consumerProperties = functionProperties.getConsumerProperties() == null
 				? new ConsumerProperties() : functionProperties.getConsumerProperties();
+		this.producerProperties = functionProperties.getProducerProperties() == null
+				? new ProducerProperties() : functionProperties.getProducerProperties();
 	}
 
 	@Override
@@ -117,6 +121,12 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
 
 	@SuppressWarnings("unchecked")
 	private <T> Message<O> toMessage(T value, Message<I> originalMessage) {
+		if(producerProperties.isUseNativeEncoding()){
+			if (logger.isDebugEnabled()) {
+				logger.debug("Native encoding enabled wrapping result to message using the original message: " + originalMessage);
+			}
+			return wrapOutputToMessage(value,originalMessage);
+		}
 		if (logger.isDebugEnabled()) {
 			logger.debug("Converting result back to message using the original message: " + originalMessage);
 		}
@@ -126,13 +136,19 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
 						: this.messageConverter.toMessage(value, originalMessage.getHeaders(), this.outputClass));
 		if (returnMessage == null) {
 			if (value.getClass().isAssignableFrom(this.outputClass)) {
-				returnMessage = (Message<O>) MessageBuilder.withPayload(value).copyHeaders(originalMessage.getHeaders()).removeHeader(MessageHeaders.CONTENT_TYPE).build();
+				returnMessage = wrapOutputToMessage(value, originalMessage);
 			}
 		}
 		Assert.notNull(returnMessage, "Failed to convert result value '" + value + "' to message.");
 		return returnMessage;
 	}
 
+	private <T> Message<O> wrapOutputToMessage(T value, Message<I> originalMessage) {
+		Message<O> returnMessage;
+		returnMessage = (Message<O>) MessageBuilder.withPayload(value).copyHeaders(originalMessage.getHeaders()).removeHeader(MessageHeaders.CONTENT_TYPE).build();
+		return returnMessage;
+	}
+
 	@SuppressWarnings("unchecked")
 	private <T> T resolveArgument(Message<I> message) {
 		if (logger.isDebugEnabled()) {

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

@@ -18,6 +18,7 @@ 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;
 
 /**
  *
@@ -35,6 +36,8 @@ public class StreamFunctionProperties {
 
 	private ConsumerProperties consumerProperties;
 
+	private ProducerProperties producerProperties;
+
 	public String getDefinition() {
 		return this.definition;
 	}
@@ -46,4 +49,16 @@ public class StreamFunctionProperties {
 	protected ConsumerProperties getConsumerProperties() {
 		return consumerProperties;
 	}
+
+	public void setConsumerProperties(ConsumerProperties consumerProperties) {
+		this.consumerProperties = consumerProperties;
+	}
+
+	protected ProducerProperties getProducerProperties() {
+		return producerProperties;
+	}
+
+	public void setProducerProperties(ProducerProperties producerProperties) {
+		this.producerProperties = producerProperties;
+	}
 }

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

@@ -21,6 +21,7 @@ import java.util.function.Function;
 
 import org.junit.Test;
 
+
 import reactor.core.publisher.Flux;
 
 import org.springframework.boot.WebApplicationType;
@@ -29,8 +30,11 @@ 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.binder.test.TestChannelBinderConfiguration;
 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.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.integration.support.MessageBuilder;
@@ -43,6 +47,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 /**
  *
  * @author Oleg Zhurakousky
+ * @author Tolga Kavukcu
  *
  */
 public class FunctionInvokerTests {
@@ -87,6 +92,62 @@ public class FunctionInvokerTests {
 		}
 	}
 
+	@Test
+	public void testNativeEncodingEnabled() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(MyFunctionsConfiguration.class))
+						.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false")) {
+
+			Message<Baz> inputMessage = new GenericMessage<>(new Baz());
+
+			StreamFunctionProperties functionProperties = createStreamFunctionPropertiesWithNativeEncoding();
+
+			functionProperties.setDefinition("pojoToPojoNonEmptyPojo");
+			FunctionInvoker<Baz, Baz> pojoToPojoSameType = new FunctionInvoker<>(functionProperties,
+					new FunctionCatalogWrapper(context.getBean(FunctionCatalog.class)),
+					context.getBean(FunctionInspector.class), context.getBean(CompositeMessageConverterFactory.class));
+			Message<Baz> outputMessage = pojoToPojoSameType.apply(Flux.just(inputMessage)).blockFirst();
+			assertThat(inputMessage.getPayload()).isEqualTo(outputMessage.getPayload());
+
+			Message<Baz> inputMessageWithBaz = new GenericMessage<>(new Baz());
+
+			functionProperties.setDefinition("messageToMessageNoType");
+			FunctionInvoker<Baz, Baz> messageToMessageNoType = new FunctionInvoker<>(functionProperties,
+					new FunctionCatalogWrapper(context.getBean(FunctionCatalog.class)), context.getBean(FunctionInspector.class), context.getBean(CompositeMessageConverterFactory.class));
+			outputMessage = messageToMessageNoType.apply(Flux.just(inputMessageWithBaz)).blockFirst();
+			assertThat(outputMessage).isInstanceOf(Message.class);
+
+			functionProperties.setDefinition("withExceptionNativeEncodingEnabled");
+			FunctionInvoker<Baz, Baz> withException = new FunctionInvoker<>(functionProperties,
+					new FunctionCatalogWrapper(context.getBean(FunctionCatalog.class)),
+					context.getBean(FunctionInspector.class), context.getBean(CompositeMessageConverterFactory.class));
+
+			Flux<Message<Baz>> fluxOfMessages = Flux.just(new GenericMessage<>(new ErrorBaz()), inputMessage);
+			Message<Baz> resultMessage = withException.apply(fluxOfMessages).blockFirst();
+			assertThat(resultMessage.getPayload()).isNotInstanceOf(ErrorFoo.class);
+		}
+	}
+
+	@Test
+	public void testWithOutNativeEncodingEnabled() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(MyFunctionsConfiguration.class))
+						.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false")) {
+
+			Message<Baz> inputMessage = new GenericMessage<>(new Baz());
+
+			StreamFunctionProperties functionProperties = createStreamFunctionProperties();
+
+			functionProperties.setDefinition("pojoToPojoNonEmptyPojo");
+			FunctionInvoker<Baz, Baz> pojoToPojoSameType = new FunctionInvoker<>(functionProperties,
+					new FunctionCatalogWrapper(context.getBean(FunctionCatalog.class)),
+					context.getBean(FunctionInspector.class), context.getBean(CompositeMessageConverterFactory.class));
+			Message<Baz> outputMessage = pojoToPojoSameType.apply(Flux.just(inputMessage)).blockFirst();
+			assertThat(inputMessage.getPayload()).isNotEqualTo(outputMessage.getPayload());
+
+		}
+	}
+
 	private StreamFunctionProperties createStreamFunctionProperties() {
 		StreamFunctionProperties functionProperties = new StreamFunctionProperties();
 		ConsumerProperties consumerProperties = new ConsumerProperties();
@@ -96,8 +157,26 @@ public class FunctionInvokerTests {
 			f.setAccessible(true);
 			f.set(functionProperties, consumerProperties);
 			return functionProperties;
+		} catch (Exception e) {
+			throw new IllegalStateException(e);
 		}
-		catch (Exception e) {
+	}
+
+	private StreamFunctionProperties createStreamFunctionPropertiesWithNativeEncoding() {
+		StreamFunctionProperties functionProperties = new StreamFunctionProperties();
+		ConsumerProperties consumerProperties = new ConsumerProperties();
+		consumerProperties.setMaxAttempts(3);
+		ProducerProperties producerProperties = new ProducerProperties();
+		producerProperties.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);
+			return functionProperties;
+		} catch (Exception e) {
 			throw new IllegalStateException(e);
 		}
 	}
@@ -130,6 +209,11 @@ public class FunctionInvokerTests {
 			return x -> x;
 		}
 
+		@Bean
+		public Function<Baz, Baz> pojoToPojoNonEmptyPojo() {
+			return x -> x;
+		}
+
 		@Bean
 		public Function<Foo, Foo> withException() {
 			return x -> {
@@ -144,6 +228,19 @@ public class FunctionInvokerTests {
 			};
 		}
 
+		@Bean
+		public Function<Baz, Baz> withExceptionNativeEncodingEnabled() {
+			return x -> {
+				if (x instanceof ErrorBaz) {
+					System.out.println("Throwing exception ");
+					throw new RuntimeException("Boom!");
+				} else {
+					System.out.println("All is good ");
+					return x;
+				}
+			};
+		}
+
 	}
 
 	private static class Foo {

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

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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.cloud.stream.function;
+
+
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.PropertySource;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ *
+ * @author Tolga Kavukcu
+ *
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@SpringBootTest(classes = { FuntionPropertiesTest.MyFunctionsConfiguration.class })
+public class FuntionPropertiesTest {
+
+	@Autowired
+	StreamFunctionProperties streamFunctionProperties;
+
+	@Test
+	public void converterCorrectlyInstalled() {
+		assertThat(streamFunctionProperties.getProducerProperties().isUseNativeEncoding()).isEqualTo(Boolean.TRUE);
+		assertThat(streamFunctionProperties.getConsumerProperties().getMaxAttempts()).isEqualTo(5);
+	}
+
+	@EnableAutoConfiguration
+	@PropertySource("classpath:/org/springframework/cloud/stream/binder/cloud-function-test.properties")
+	@Configuration
+	public static class MyFunctionsConfiguration {
+
+		@Bean
+		public StreamFunctionProperties streamFunctionProperties() {
+			return new StreamFunctionProperties();
+		}
+	}
+}

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

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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.cloud.stream.function.pojo;
+
+
+/**
+ * Serializable pojo with object mapper.
+ * @author Tolga Kavukcu
+ */
+public class Baz {
+	private String baz = "baz";
+
+	public String getBaz() {
+		return baz;
+	}
+
+	public void setBaz(String baz) {
+		this.baz = baz;
+	}
+}

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

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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.cloud.stream.function.pojo;
+
+
+/**
+ * Serializable pojo with object mapper.
+ * @author Tolga Kavukcu
+ */
+public class ErrorBaz extends Baz{
+	private String baz = "bazError";
+
+	public String getBaz() {
+		return baz;
+	}
+
+	public void setBaz(String baz) {
+		this.baz = baz;
+	}
+}

spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/cloud-function-test.properties

@@ -0,0 +1,2 @@
+spring.cloud.stream.function.producerProperties.useNativeEncoding: true
+spring.cloud.stream.function.consumerProperties.maxAttempts: 5