diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionCatalog.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionCatalog.java index bda1f2393..4147263a4 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionCatalog.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionCatalog.java @@ -17,6 +17,9 @@ package org.springframework.cloud.function.context; import java.util.Set; +import java.util.function.Function; + +import org.springframework.cloud.function.context.catalog.OutputPostProcessor; /** * @author Dave Syer @@ -24,6 +27,11 @@ import java.util.Set; */ public interface FunctionCatalog { + + default T lookup(String name, OutputPostProcessor outputPostProcessor) { + throw new UnsupportedOperationException("This type of lookup is not supported by this instance of FunctionCatalog"); + } + /** * Will look up the instance of the functional interface by name only. * @param instance type diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeConversionHelper.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeConversionHelper.java index 158f8deb0..58771dcfa 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeConversionHelper.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeConversionHelper.java @@ -58,33 +58,33 @@ class FunctionTypeConversionHelper { @SuppressWarnings("rawtypes") - Object convertInput(Object input) { + Object convertInputIfNecessary(Object input) { List convertedResults = new ArrayList(); if (input instanceof Tuple2) { - convertedResults.add(this.doConvertInput(((Tuple2)input).getT1(), getInputArgumentType(0))); - convertedResults.add(this.doConvertInput(((Tuple2)input).getT2(), getInputArgumentType(1))); + convertedResults.add(this.doConvert(((Tuple2)input).getT1(), getInputArgumentType(0))); + convertedResults.add(this.doConvert(((Tuple2)input).getT2(), getInputArgumentType(1))); } if (input instanceof Tuple3) { - convertedResults.add(this.doConvertInput(((Tuple3)input).getT3(), getInputArgumentType(2))); + convertedResults.add(this.doConvert(((Tuple3)input).getT3(), getInputArgumentType(2))); } if (input instanceof Tuple4) { - convertedResults.add(this.doConvertInput(((Tuple4)input).getT4(), getInputArgumentType(3))); + convertedResults.add(this.doConvert(((Tuple4)input).getT4(), getInputArgumentType(3))); } if (input instanceof Tuple5) { - convertedResults.add(this.doConvertInput(((Tuple5)input).getT5(), getInputArgumentType(4))); + convertedResults.add(this.doConvert(((Tuple5)input).getT5(), getInputArgumentType(4))); } if (input instanceof Tuple6) { - convertedResults.add(this.doConvertInput(((Tuple6)input).getT6(), getInputArgumentType(5))); + convertedResults.add(this.doConvert(((Tuple6)input).getT6(), getInputArgumentType(5))); } if (input instanceof Tuple7) { - convertedResults.add(this.doConvertInput(((Tuple7)input).getT7(), getInputArgumentType(6))); + convertedResults.add(this.doConvert(((Tuple7)input).getT7(), getInputArgumentType(6))); } if (input instanceof Tuple8) { - convertedResults.add(this.doConvertInput(((Tuple8)input).getT8(), getInputArgumentType(7))); + convertedResults.add(this.doConvert(((Tuple8)input).getT8(), getInputArgumentType(7))); } input = CollectionUtils.isEmpty(convertedResults) - ? this.doConvertInput(input, getInputArgumentType(0)) + ? this.doConvert(input, getInputArgumentType(0)) : Tuples.fromArray(convertedResults.toArray()); return input; } @@ -114,6 +114,17 @@ class FunctionTypeConversionHelper { } } + Type getOutputArgumentType(int index) { + if (this.functionArgumentTypes[1] instanceof ParameterizedType) { + Type[] types = ((ParameterizedType)this.functionArgumentTypes[1]).getActualTypeArguments(); + + return (types[index]);//.getActualTypeArguments()[0]; + } + else { + return this.functionArgumentTypes[1]; + } + } + private Class getRawType(Type targetType) { if (targetType instanceof ParameterizedType && Publisher.class.isAssignableFrom((Class)((ParameterizedType)targetType).getRawType())) { targetType = ((ParameterizedType)targetType).getActualTypeArguments()[0]; @@ -125,14 +136,14 @@ class FunctionTypeConversionHelper { } @SuppressWarnings({ "unchecked", "rawtypes" }) - private Object doConvertInput(Object input, Type targetType) { + private Object doConvert(Object input, Type targetType) { Class actualInputType = this.getRawType(targetType); if (input instanceof Publisher) { if (!actualInputType.isAssignableFrom(Void.class)) { - input = input instanceof Flux - ? ((Flux) input).map(value -> this.convertInputArgument(value, targetType, actualInputType)) - : ((Mono) input).map(value -> this.convertInputArgument(value, targetType, actualInputType)); + input = input instanceof Mono + ? Mono.from((Publisher) input).map(value -> this.convertInputArgument(value, targetType, actualInputType)) + : Flux.from((Publisher) input).map(value -> this.convertInputArgument(value, targetType, actualInputType)); } } else { diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistry.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistry.java index d91b54054..5adcb016e 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistry.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistry.java @@ -19,6 +19,7 @@ package org.springframework.cloud.function.context.catalog; import java.util.HashMap; import java.util.Map; import java.util.Set; +import java.util.function.BiFunction; import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; @@ -45,6 +46,7 @@ import org.springframework.util.StringUtils; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.util.function.Tuple2; /** * @@ -70,10 +72,16 @@ public class LazyFunctionRegistry implements FunctionRegistry, FunctionInspector this.messageConverter = messageConverter; } + @SuppressWarnings("unchecked") + @Override + public T lookup(String definition, OutputPostProcessor outputPostProcessor) { + return (T) this.compose(null, definition, false, outputPostProcessor); + } + @SuppressWarnings("unchecked") @Override public T lookup(Class type, String definition) { - return (T) this.compose(type, definition); + return (T) this.compose(type, definition, false, new DefaultOutputPostProcessor()); } @Override @@ -146,13 +154,24 @@ public class LazyFunctionRegistry implements FunctionRegistry, FunctionInspector // } // } // return key; +// } + + private boolean notStandardFunction(Object function) { + return !(function instanceof Supplier || function instanceof Function || function instanceof Consumer); + } + +// private Object toCannonical(Object function) { +// if (function instanceof BiFunction) { +// return new Function>() { +// }; +// } // } @SuppressWarnings({ "unchecked", "rawtypes" }) - private Function compose(Class type, String definition) { + private Function compose(Class type, String definition, boolean raw, OutputPostProcessor outputPostProcessor) { Function resultFunction = null; if (this.registrationsByName.containsKey(definition)) { - resultFunction = new FunctionInvocationWrapper(this.registrationsByName.get(definition), false); + resultFunction = new FunctionInvocationWrapper(this.registrationsByName.get(definition), false, outputPostProcessor); } else { String[] names = StringUtils.delimitedListToStringArray(definition.replaceAll(",", "|").trim(), "|"); @@ -168,6 +187,11 @@ public class LazyFunctionRegistry implements FunctionRegistry, FunctionInspector composedNameBuilder.append(name); Object function = this.applicationContext.getBean(name); + +// if (notStandardFunction(function) && type != null && Function.class.isAssignableFrom(type)) { +// +// } + FunctionType funcType = beanDefinitionExists(name) ? FunctionType.of(FunctionContextUtils.findType(name, (ConfigurableListableBeanFactory) applicationContext.getBeanFactory())) : new FunctionType(function.getClass()); @@ -176,7 +200,7 @@ public class LazyFunctionRegistry implements FunctionRegistry, FunctionInspector FunctionRegistration registration = new FunctionRegistration<>(function, name).type(funcType); registrationsByFunction.putIfAbsent(function, registration); registrationsByName.putIfAbsent(name, registration); - function = new FunctionInvocationWrapper(registration, false); + function = new FunctionInvocationWrapper(registration, false, outputPostProcessor); if (resultFunction == null) { resultFunction = (Function) function; } @@ -197,7 +221,7 @@ public class LazyFunctionRegistry implements FunctionRegistry, FunctionInspector registration = new FunctionRegistration(resultFunction, composedNameBuilder.toString()).type(funcType); registrationsByFunction.putIfAbsent(resultFunction, registration); registrationsByName.putIfAbsent(composedNameBuilder.toString(), registration); - resultFunction = new FunctionInvocationWrapper(registration, true); + resultFunction = new FunctionInvocationWrapper(registration, true, outputPostProcessor); } previousFunctionType = funcType; prefix = "|"; @@ -228,10 +252,13 @@ public class LazyFunctionRegistry implements FunctionRegistry, FunctionInspector private final FunctionTypeConversionHelper functionTypeConversionHelper; - FunctionInvocationWrapper(FunctionRegistration functionRegistration, boolean composed) { + private final OutputPostProcessor outputPostProcessor; + + FunctionInvocationWrapper(FunctionRegistration functionRegistration, boolean composed, OutputPostProcessor outputPostProcessor) { this.target = functionRegistration.getTarget(); this.functionRegistration = functionRegistration; this.composed = composed; + this.outputPostProcessor = outputPostProcessor; this.functionTypeConversionHelper = new FunctionTypeConversionHelper(this.functionRegistration, conversionService, messageConverter); } @@ -249,10 +276,10 @@ public class LazyFunctionRegistry implements FunctionRegistry, FunctionInspector @Override public Object get() { // wrap/unwrap to/from reactive - Object input = Flux.class.isAssignableFrom(this.functionRegistration.getType().getInputWrapper()) - ? Flux.empty() - : (Mono.class.isAssignableFrom(this.functionRegistration.getType().getInputWrapper()) - ? Mono.empty() + Object input = Mono.class.isAssignableFrom(this.functionRegistration.getType().getInputWrapper()) + ? Mono.empty() + : (Flux.class.isAssignableFrom(this.functionRegistration.getType().getInputWrapper()) + ? Flux.empty() : null); return this.doApply(input, false); @@ -266,10 +293,10 @@ public class LazyFunctionRegistry implements FunctionRegistry, FunctionInspector input = this.wrapInputToReactiveIfNecessary(input); - Object result = null; + Object result; if (input instanceof Publisher) { if (!this.composed) { - input = this.functionTypeConversionHelper.convertInput(input); + input = this.functionTypeConversionHelper.convertInputIfNecessary(input); } result = this.applyReactive((Publisher) input, consumer); } @@ -280,12 +307,14 @@ public class LazyFunctionRegistry implements FunctionRegistry, FunctionInspector } else { if (!this.composed) { - input = this.functionTypeConversionHelper.convertInput(input); + input = this.functionTypeConversionHelper.convertInputIfNecessary(input); } result = this.applyImperative(input, consumer); } } + result = this.outputPostProcessor.postProcessOutput(result); + return this.wrapOutputToReactiveIfNecessary(result); } @@ -354,8 +383,9 @@ public class LazyFunctionRegistry implements FunctionRegistry, FunctionInspector result = publisher instanceof Mono ? Mono.just(result) : Flux.just(result); } else { - result = publisher instanceof Flux ? Flux.from(publisher).map(value -> ((Function)this.target).apply(value)) - : Mono.from(publisher).map(value -> ((Function)this.target).apply(value)); + result = publisher instanceof Mono + ? Mono.from(publisher).map(value -> ((Function)this.target).apply(value)) + : Flux.from(publisher).map(value -> ((Function)this.target).apply(value)); } } } 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 c7da5d0a3..ab565fa15 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 @@ -17,9 +17,11 @@ package org.springframework.cloud.function.context.config; import java.lang.reflect.Type; +import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.LinkedHashSet; @@ -33,9 +35,6 @@ import java.util.stream.Stream; import javax.annotation.PreDestroy; -import com.fasterxml.jackson.databind.ObjectMapper; -import com.google.gson.Gson; - import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; @@ -67,6 +66,9 @@ import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.FilterType; import org.springframework.core.annotation.AnnotatedElementUtils; import org.springframework.core.convert.ConversionService; +import org.springframework.core.convert.TypeDescriptor; +import org.springframework.core.convert.converter.ConditionalGenericConverter; +import org.springframework.core.convert.support.ConfigurableConversionService; import org.springframework.core.convert.support.DefaultConversionService; import org.springframework.core.type.StandardMethodMetadata; import org.springframework.lang.Nullable; @@ -75,6 +77,10 @@ import org.springframework.messaging.converter.CompositeMessageConverter; import org.springframework.messaging.converter.MappingJackson2MessageConverter; import org.springframework.messaging.converter.MessageConverter; import org.springframework.messaging.converter.StringMessageConverter; +import org.springframework.util.ClassUtils; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.gson.Gson; /** * @author Dave Syer @@ -106,9 +112,35 @@ public class ContextFunctionCatalogAutoConfiguration { messageConverters.add(new ByteArrayMessageConverter()); messageConverter = new CompositeMessageConverter(messageConverters); } + if (conversionService != null) { + ((ConfigurableConversionService)conversionService).addConverter(new MyConverter()); + } return new LazyFunctionRegistry(conversionService, messageConverter); } + public static class MyConverter implements ConditionalGenericConverter { + + @Override + public Set getConvertibleTypes() { + return Collections.singleton(new ConvertiblePair(byte[].class, String.class)); + } + + @Override + public Object convert(Object source, TypeDescriptor sourceType, TypeDescriptor targetType) { + return new String(((byte[])source), StandardCharsets.UTF_8); + } + + @Override + public boolean matches(TypeDescriptor sourceType, TypeDescriptor targetType) { + if (ClassUtils.isAssignable(sourceType.getType(), byte[].class) && ClassUtils.isAssignable(targetType.getType(), String.class)) { + // maybe + return true; + } + return false; + } + + } + @Bean(RoutingFunction.FUNCTION_NAME) @ConditionalOnProperty(name = "spring.cloud.function.routing.enabled", havingValue = "true") RoutingFunction gateway(FunctionCatalog functionCatalog, FunctionInspector functionInspector) { diff --git a/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistryMultiInOutTests.java b/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistryMultiInOutTests.java index a78076a32..62fc08773 100644 --- a/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistryMultiInOutTests.java +++ b/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistryMultiInOutTests.java @@ -19,6 +19,7 @@ package org.springframework.cloud.function.context.catalog; import java.nio.charset.StandardCharsets; import java.util.Collections; import java.util.List; +import java.util.function.BiFunction; import java.util.function.Function; import org.junit.Test; @@ -68,6 +69,19 @@ public class LazyFunctionRegistryMultiInOutTests { System.out.println(result); } + @SuppressWarnings("unused") + @Test + public void testMultiInputBiFunction() { + FunctionCatalog catalog = this.configureCatalog(); + BiFunction, Flux, Flux> multiInputFunction = + catalog.lookup(BiFunction.class, "multiInputSingleOutputViaBiFunction"); + Flux stringStream = Flux.just("one", "two", "three"); + Flux intStream = Flux.just(1, 2, 3); + +// List result = multiInputFunction.apply(Tuples.of(stringStream, intStream)).collectList().block(); +// System.out.println(result); + } + /* * This test invokes the same function as above but with types reversed. * While the target function remains , Flux> @@ -206,6 +220,21 @@ public class LazyFunctionRegistryMultiInOutTests { result.getT2().subscribe(v -> System.out.println("=> 2: " + v)); } +// @Test +// public void testMultiToMultiByteArray() { +// FunctionCatalog catalog = this.configureCatalog(); +// Function, Flux, Flux>, Tuple2, Mono>> multiTuMulti = +// catalog.lookup("multiTuMulti"); +// +// Flux firstFlux = Flux.just("Unlce", "Oncle"); +// Flux secondFlux = Flux.just("Sam", "Pierre"); +// Flux thirdFlux = Flux.just(1, 2); +// +// Tuple2, Mono> result = multiTuMulti.apply(Tuples.of(firstFlux, secondFlux, thirdFlux)); +// result.getT1().subscribe(v -> System.out.println("=> 1: " + v)); +// result.getT2().subscribe(v -> System.out.println("=> 2: " + v)); +// } + @EnableAutoConfiguration @Configuration @@ -227,6 +256,15 @@ public class LazyFunctionRegistryMultiInOutTests { }; } + @Bean + public BiFunction, Flux, Flux> multiInputSingleOutputViaBiFunction() { + return (in1, in2) -> { + Flux stringStream = in1; + Flux intStream = in2; + return Flux.zip(stringStream, intStream, (string, integer) -> string + "-" + integer); + }; + } + @Bean public Function, Tuple3, Flux, Flux>> multiOutputAsTuplePojoIn(){ return flux -> { diff --git a/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistryTests.java b/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistryTests.java index a2579885d..017355481 100644 --- a/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistryTests.java +++ b/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistryTests.java @@ -16,6 +16,7 @@ package org.springframework.cloud.function.context.catalog; +import java.lang.reflect.ParameterizedType; import java.util.List; import java.util.function.Consumer; import java.util.function.Function; @@ -28,6 +29,7 @@ import reactor.util.function.Tuple2; import reactor.util.function.Tuple3; import reactor.util.function.Tuples; +import org.hamcrest.internal.ReflectiveTypeFinder; import org.junit.Test; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; @@ -67,6 +69,17 @@ public class LazyFunctionRegistryTests { assertThat(result.get(1)).isEqualTo("UPPERCASEFLUX2"); } + + @Test + public void testSerializationDeserialization() { + FunctionCatalog catalog = this.configureCatalog(); + + //Function asIs = catalog.lookup("uppercase", new ); + + //ParameterizedType +// + } + /* * When invoking imperative function as reactive the rules are * - the input wrapper must match the output wrapper (e.g., or ) @@ -191,7 +204,15 @@ public class LazyFunctionRegistryTests { List result = multiInputFunction.apply(Tuples.of(stringStream, intStream)).collectList().block(); System.out.println(result); } - + + + private class MyFunction { + public String apply(String a, Integer b, Foo c) { + + } + } + + @Test public void testMultiInputWithComposition() { FunctionCatalog catalog = this.configureCatalog();