From 6ffae9397a9e3e883637a5a0d70501388aa8fd7e Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Thu, 6 Jun 2024 11:38:19 +0200 Subject: [PATCH 1/3] GH-1141 Add support for composing reactive Supplier/Function with imperative Consumer Resolves #1141 --- .../catalog/SimpleFunctionRegistry.java | 30 +++++++- ...BeanFactoryAwareFunctionRegistryTests.java | 74 +++++++++++++++++++ 2 files changed, 101 insertions(+), 3 deletions(-) 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 252b52389..cba41a6a4 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 @@ -394,7 +394,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry { @SuppressWarnings("rawtypes") public class FunctionInvocationWrapper implements Function, Consumer, Supplier, Runnable { - private final Object target; + private Object target; private Type inputType; @@ -658,13 +658,21 @@ public class SimpleFunctionRegistry implements FunctionRegistry { || FunctionTypeUtils.isMultipleArgumentType(((FunctionInvocationWrapper) after).outputType)) { throw new UnsupportedOperationException("Composition of functions with multiple arguments is not supported at the moment"); } + FunctionInvocationWrapper afterWrapper = (FunctionInvocationWrapper) after; + + //see GH-1141 for this code snippet + if ((this.getTarget() instanceof Supplier || this.getTarget() instanceof Function) && FunctionTypeUtils.isPublisher(this.getOutputType()) + && afterWrapper.getTarget() instanceof Consumer && !FunctionTypeUtils.isPublisher(afterWrapper.getInputType())) { + Consumer wrapper = new ConsumerWrapper((Consumer) afterWrapper.getTarget()); + afterWrapper.target = wrapper; + afterWrapper.inputType = this.outputType; + } + // this.setSkipOutputConversion(true); ((FunctionInvocationWrapper) after).setSkipOutputConversion(true); Function rawComposedFunction = v -> ((FunctionInvocationWrapper) after).doApply(doApply(v)); - FunctionInvocationWrapper afterWrapper = (FunctionInvocationWrapper) after; - Type composedFunctionType; if (afterWrapper.outputType == null) { composedFunctionType = (this.inputType == null) ? @@ -1551,4 +1559,20 @@ public class SimpleFunctionRegistry implements FunctionRegistry { return t; } } + + @SuppressWarnings({ "unchecked", "rawtypes" }) + private static class ConsumerWrapper implements Consumer> { + + private final Consumer targetConsumer; + + ConsumerWrapper(Consumer targetConsumer) { + this.targetConsumer = targetConsumer; + } + + @Override + public void accept(Flux messageFlux) { + messageFlux.doOnNext(this.targetConsumer).subscribe(); + } + + } } diff --git a/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistryTests.java b/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistryTests.java index edd78aeac..9407a1c11 100644 --- a/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistryTests.java +++ b/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistryTests.java @@ -140,6 +140,37 @@ public class BeanFactoryAwareFunctionRegistryTests { assertThat(registration.getNames().iterator().next()).isEqualTo("echo1"); } + @Test + public void testCompositionReactiveSupplierWithImplicitConsumer() throws Exception { + FunctionCatalog catalog = this.configureCatalog(CompositionReactiveSupplierWithConsumer.class); + FunctionInvocationWrapper function = catalog.lookup("supplyPrimitive|consume"); + function.apply(null); + assertThat(CompositionReactiveSupplierWithConsumer.results.size()).isEqualTo(2); + assertThat(CompositionReactiveSupplierWithConsumer.results.get(0)).isEqualTo(1); + assertThat(CompositionReactiveSupplierWithConsumer.results.get(1)).isEqualTo(2); + CompositionReactiveSupplierWithConsumer.results.clear(); + + function = catalog.lookup("supplyMessage|consume"); + function.apply(null); + assertThat(CompositionReactiveSupplierWithConsumer.results.size()).isEqualTo(2); + assertThat(CompositionReactiveSupplierWithConsumer.results.get(0)).isEqualTo(1); + assertThat(CompositionReactiveSupplierWithConsumer.results.get(1)).isEqualTo(2); + CompositionReactiveSupplierWithConsumer.results.clear(); + + function = catalog.lookup("functionMessage|consume"); + function.apply(Flux.fromArray(new Message[] {MessageBuilder.withPayload("ricky").build(), MessageBuilder.withPayload("bubbles").build()})); + assertThat(CompositionReactiveSupplierWithConsumer.results.size()).isEqualTo(2); + assertThat(CompositionReactiveSupplierWithConsumer.results.get(0)).isEqualTo("RICKY"); + assertThat(CompositionReactiveSupplierWithConsumer.results.get(1)).isEqualTo("BUBBLES"); + CompositionReactiveSupplierWithConsumer.results.clear(); + + function = catalog.lookup("functionPrimitive|consume"); + function.apply(Flux.fromArray(new String[] {"ricky", "bubbles"})); + assertThat(CompositionReactiveSupplierWithConsumer.results.size()).isEqualTo(2); + assertThat(CompositionReactiveSupplierWithConsumer.results.get(0)).isEqualTo("RICKY"); + assertThat(CompositionReactiveSupplierWithConsumer.results.get(1)).isEqualTo("BUBBLES"); + } + @SuppressWarnings({ "rawtypes", "unchecked" }) @Test public void testMessageWithArrayAsPayload() throws Exception { @@ -1539,6 +1570,49 @@ public class BeanFactoryAwareFunctionRegistryTests { } } + @EnableAutoConfiguration + @Configuration // s-c-f-1141 + @SuppressWarnings({"unchecked", "rawtypes"}) + public static class CompositionReactiveSupplierWithConsumer { + private static List results = new ArrayList<>(); + + @Bean + public Function, Flux> functionPrimitive() { + return flux -> flux.map(v -> v.toUpperCase()); + } + + @Bean + public Function>, Flux>> functionMessage() { + return flux -> flux.map(v -> MessageBuilder.withPayload(v.getPayload().toUpperCase()).build()); + } + + @Bean + public Supplier>> supplyMessage() { + return () -> { + return Flux.fromArray( + new Message[] { MessageBuilder.withPayload(1).build(), MessageBuilder.withPayload(2).build() }); + }; + } + + @Bean + public Supplier> supplyPrimitive() { + return () -> { + return Flux.fromArray( + new Integer[] { 1, 2}); + }; + } + + @Bean + public Consumer consume() { + return v -> { + if (v instanceof Message vMessage) { + v = vMessage.getPayload(); + } + results.add(v); + }; + } + } + @EnableAutoConfiguration @Configuration public static class MessageWithArrayAsPayload { From b17795dd66cafc19c0809b137b704a5c27a091f5 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Fri, 7 Jun 2024 07:54:51 +0200 Subject: [PATCH 2/3] Revert "GH-1148 Remove manual setting of ObjectMapper feature" This reverts commit 812c39eccba900a8d4bedef963e65983eae65e53. --- .../context/config/ContextFunctionCatalogAutoConfiguration.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java index e5371b35c..1d0b50871 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java @@ -24,6 +24,7 @@ import java.util.function.Function; import java.util.function.Supplier; import java.util.stream.Collectors; +import com.fasterxml.jackson.databind.DeserializationFeature; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.SerializationFeature; import com.google.gson.Gson; @@ -221,6 +222,7 @@ public class ContextFunctionCatalogAutoConfiguration { mapper = new ObjectMapper(); } mapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false); + mapper.configure(DeserializationFeature.FAIL_ON_TRAILING_TOKENS, true); return new JacksonMapper(mapper); } } From 8b66fd296e72e8c0f1ed5df0711da5ea75527cb8 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Fri, 7 Jun 2024 08:02:59 +0200 Subject: [PATCH 3/3] GH-1148 Configure our own ObjectMapper --- .../config/ContextFunctionCatalogAutoConfiguration.java | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java index 1d0b50871..39732daed 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java @@ -214,13 +214,7 @@ public class ContextFunctionCatalogAutoConfiguration { } private JsonMapper jackson(ApplicationContext context) { - ObjectMapper mapper; - try { - mapper = context.getBean(ObjectMapper.class); - } - catch (Exception e) { - mapper = new ObjectMapper(); - } + ObjectMapper mapper = new ObjectMapper(); mapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false); mapper.configure(DeserializationFeature.FAIL_ON_TRAILING_TOKENS, true); return new JacksonMapper(mapper);