diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java index eeaa7b6ee..35ea8d448 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java +++ b/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 implements Function>, Flux implements Function>, Flux implements Function>, Flux Message toMessage(T value, Message 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 implements Function>, Flux) 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 Message wrapOutputToMessage(T value, Message originalMessage) { + Message returnMessage; + returnMessage = (Message) MessageBuilder.withPayload(value).copyHeaders(originalMessage.getHeaders()).removeHeader(MessageHeaders.CONTENT_TYPE).build(); + return returnMessage; + } + @SuppressWarnings("unchecked") private T resolveArgument(Message message) { if (logger.isDebugEnabled()) { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java index 0fe2be729..54565fa48 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java +++ b/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; + } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java index d53631fff..a333067ad 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java +++ b/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 inputMessage = new GenericMessage<>(new Baz()); + + StreamFunctionProperties functionProperties = createStreamFunctionPropertiesWithNativeEncoding(); + + functionProperties.setDefinition("pojoToPojoNonEmptyPojo"); + FunctionInvoker pojoToPojoSameType = new FunctionInvoker<>(functionProperties, + new FunctionCatalogWrapper(context.getBean(FunctionCatalog.class)), + context.getBean(FunctionInspector.class), context.getBean(CompositeMessageConverterFactory.class)); + Message outputMessage = pojoToPojoSameType.apply(Flux.just(inputMessage)).blockFirst(); + assertThat(inputMessage.getPayload()).isEqualTo(outputMessage.getPayload()); + + Message inputMessageWithBaz = new GenericMessage<>(new Baz()); + + functionProperties.setDefinition("messageToMessageNoType"); + FunctionInvoker 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 withException = new FunctionInvoker<>(functionProperties, + new FunctionCatalogWrapper(context.getBean(FunctionCatalog.class)), + context.getBean(FunctionInspector.class), context.getBean(CompositeMessageConverterFactory.class)); + + Flux> fluxOfMessages = Flux.just(new GenericMessage<>(new ErrorBaz()), inputMessage); + Message 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 inputMessage = new GenericMessage<>(new Baz()); + + StreamFunctionProperties functionProperties = createStreamFunctionProperties(); + + functionProperties.setDefinition("pojoToPojoNonEmptyPojo"); + FunctionInvoker pojoToPojoSameType = new FunctionInvoker<>(functionProperties, + new FunctionCatalogWrapper(context.getBean(FunctionCatalog.class)), + context.getBean(FunctionInspector.class), context.getBean(CompositeMessageConverterFactory.class)); + Message 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 pojoToPojoNonEmptyPojo() { + return x -> x; + } + @Bean public Function withException() { return x -> { @@ -144,6 +228,19 @@ public class FunctionInvokerTests { }; } + @Bean + public Function 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 { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FuntionPropertiesTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FuntionPropertiesTest.java new file mode 100644 index 000000000..321dbf2e5 --- /dev/null +++ b/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(); + } + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/pojo/Baz.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/pojo/Baz.java new file mode 100644 index 000000000..d2e907c12 --- /dev/null +++ b/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; + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/pojo/ErrorBaz.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/pojo/ErrorBaz.java new file mode 100644 index 000000000..17443da1b --- /dev/null +++ b/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; + } +} diff --git a/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/cloud-function-test.properties b/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/cloud-function-test.properties new file mode 100644 index 000000000..2f517c8e6 --- /dev/null +++ b/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