diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java index e470d1b5e..c2233d09e 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java @@ -636,6 +636,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry { @Override public Function andThen(Function after) { Assert.isTrue(after instanceof FunctionInvocationWrapper, "Composed function must be an instanceof FunctionInvocationWrapper."); + if (FunctionTypeUtils.isMultipleArgumentType(this.inputType) || FunctionTypeUtils.isMultipleArgumentType(this.outputType) || FunctionTypeUtils.isMultipleArgumentType(((FunctionInvocationWrapper) after).inputType) @@ -643,6 +644,8 @@ public class SimpleFunctionRegistry implements FunctionRegistry { throw new UnsupportedOperationException("Composition of functions with multiple arguments is not supported at the moment"); } + this.setSkipOutputConversion(true); + ((FunctionInvocationWrapper) after).setSkipOutputConversion(true); Function rawComposedFunction = v -> ((FunctionInvocationWrapper) after).doApply(doApply(v)); FunctionInvocationWrapper afterWrapper = (FunctionInvocationWrapper) after; diff --git a/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistryTests.java b/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistryTests.java index db2a902c9..fe012044a 100644 --- a/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistryTests.java +++ b/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistryTests.java @@ -59,6 +59,7 @@ import org.springframework.cloud.function.json.JacksonMapper; import org.springframework.cloud.function.json.JsonMapper; import org.springframework.context.ApplicationContext; import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; import org.springframework.core.ResolvableType; import org.springframework.core.convert.ConversionService; import org.springframework.core.convert.support.DefaultConversionService; @@ -550,6 +551,14 @@ public class SimpleFunctionRegistryTests { assertThat(FunctionTypeUtils.isMono(function.getOutputType())); } + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Test + public void testHeaderPropagationInComposedFunction() { + FunctionCatalog catalog = this.configureCatalog(GH_1063_Configuration.class); + Consumer function = catalog.lookup("uppercase|reverse|print"); + function.accept("hello"); + } + @Test public void testFunctionCompositionWithReactiveSupplierAndConsumer() { SimpleFunctionRegistry catalog = new SimpleFunctionRegistry(this.conversionService, this.messageConverter, @@ -820,4 +829,24 @@ public class SimpleFunctionRegistryTests { } } + @EnableAutoConfiguration + @Configuration + public static class GH_1063_Configuration { + + @Bean + Function> uppercase() { + return input -> MessageBuilder.withPayload(input).setHeader("FOO", "BAR").build(); + } + + @Bean + Function reverse() { + return payload -> new StringBuilder(payload).reverse().toString(); + } + + @Bean + Consumer> print() { + return msg -> assertThat(msg.getHeaders().get("FOO")).isEqualTo("BAR"); + } + } + }