SpringCloud/spring-cloud-function
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

interim

Browse Source
This commit is contained in:
Oleg Zhurakousky
2019-06-17 15:16:17 +02:00
parent fe91201561
commit 98bdd8237a
6 changed files with 173 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionCatalog.java
View File

@@ -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> 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 <T> instance type

39
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeConversionHelper.java
View File

@@ -58,33 +58,33 @@ class FunctionTypeConversionHelper {
@SuppressWarnings("rawtypes")
Object convertInput(Object input) {
Object convertInputIfNecessary(Object input) {
List<Object> convertedResults = new ArrayList<Object>();
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 {

60
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistry.java
View File

@@ -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> T lookup(String definition, OutputPostProcessor outputPostProcessor) {
return (T) this.compose(null, definition, false, outputPostProcessor);
}
@SuppressWarnings("unchecked")
@Override
public <T> 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<Tuple2<T1, T2>>() {
// };
// }
// }
@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<Object> 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<Object>(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<Object>) 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));
}
}
}

38
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java
View File

@@ -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<ConvertiblePair> 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) {

38
spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistryMultiInOutTests.java
View File

@@ -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<String>, Flux<Integer>, Flux<String>> multiInputFunction =
catalog.lookup(BiFunction.class, "multiInputSingleOutputViaBiFunction");
Flux<String> stringStream = Flux.just("one", "two", "three");
Flux<Integer> intStream = Flux.just(1, 2, 3);
// List<String> 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 <Tuple2<Flux<String>, Flux<Integer>>
@@ -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<Tuple3<Flux<String>, Flux<String>, Flux<Integer>>, Tuple2<Flux<Person>, Mono<Long>>> multiTuMulti =
// catalog.lookup("multiTuMulti");
//
// Flux<String> firstFlux = Flux.just("Unlce", "Oncle");
// Flux<String> secondFlux = Flux.just("Sam", "Pierre");
// Flux<Integer> thirdFlux = Flux.just(1, 2);
//
// Tuple2<Flux<Person>, Mono<Long>> 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<String>, Flux<Integer>, Flux<String>> multiInputSingleOutputViaBiFunction() {
return (in1, in2) -> {
Flux<String> stringStream = in1;
Flux<Integer> intStream = in2;
return Flux.zip(stringStream, intStream, (string, integer) -> string + "-" + integer);
};
}
@Bean
public Function<Flux<Person>, Tuple3<Flux<Person>, Flux<String>, Flux<Integer>>> multiOutputAsTuplePojoIn(){
return flux -> {

23
spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/LazyFunctionRegistryTests.java
View File

@@ -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<byte[], byte[]> 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., <Flux, Flux> or <Mono, Mono>)
@@ -191,7 +204,15 @@ public class LazyFunctionRegistryTests {
List<String> 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();