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

Merge branch '4.x' into main

Browse Source
This commit is contained in:
Oleg Zhurakousky
2022-01-17 15:13:34 +01:00
parent ff9cb57741 5074b7eee4
commit a1a3c236d0
116 changed files with 521 additions and 5557 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
spring-cloud-function-context/pom.xml
View File

@@ -12,7 +12,7 @@
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-function-parent</artifactId>
<version>3.2.2-SNAPSHOT</version>
<version>4.0.0-SNAPSHOT</version>
</parent>
<properties>
@@ -46,6 +46,19 @@
<groupId>com.fasterxml.jackson.module</groupId>
<artifactId>jackson-module-kotlin</artifactId>
<optional>true</optional>
<<<<<<< HEAD
=======
</dependency>
<dependency>
<groupId>javax.annotation</groupId>
<artifactId>javax.annotation-api</artifactId>
<version>1.3.2</version>
</dependency>
<dependency>
<groupId>javax.activation</groupId>
<artifactId>javax.activation-api</artifactId>
<version>1.2.0</version>
>>>>>>> 4.x
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>

4
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/AbstractSpringFunctionAdapterInitializer.java
View File

@@ -141,7 +141,7 @@ public abstract class AbstractSpringFunctionAdapterInitializer<C> implements Clo
return FunctionTypeUtils.getRawType(FunctionTypeUtils.getGenericType(((FunctionInvocationWrapper) func).getInputType()));
}
if (functionRegistration != null) {
return functionRegistration.getType().getInputType();
return FunctionTypeUtils.getRawType(FunctionTypeUtils.getInputType(functionRegistration.getType()));
}
return Object.class;
}
@@ -287,7 +287,7 @@ public abstract class AbstractSpringFunctionAdapterInitializer<C> implements Clo
Type type = FunctionContextUtils.
findType(name, this.context.getBeanFactory());
this.functionRegistration = functionRegistration.type(new FunctionType(type));
this.functionRegistration = functionRegistration.type(type);
((FunctionRegistry) this.catalog).register(functionRegistration);
return this.catalog.lookup(name);

144
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionRegistration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-2021 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.
@@ -27,27 +27,13 @@ import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import net.jodah.typetools.TypeResolver;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.cloud.function.context.catalog.FunctionTypeUtils;
import org.springframework.cloud.function.context.config.RoutingFunction;
import org.springframework.cloud.function.core.FluxConsumer;
import org.springframework.cloud.function.core.FluxFunction;
import org.springframework.cloud.function.core.FluxSupplier;
import org.springframework.cloud.function.core.FluxToMonoFunction;
import org.springframework.cloud.function.core.FluxedConsumer;
import org.springframework.cloud.function.core.FluxedFunction;
import org.springframework.cloud.function.core.MonoSupplier;
import org.springframework.cloud.function.core.MonoToFluxFunction;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
/**
* @param <T> target type
* @author Dave Syer
@@ -71,7 +57,7 @@ public class FunctionRegistration<T> implements BeanNameAware {
private T target;
private FunctionType type;
private Type type;
/**
* Creates instance of FunctionRegistration.
@@ -105,7 +91,7 @@ public class FunctionRegistration<T> implements BeanNameAware {
this.names.addAll(names);
}
public FunctionType getType() {
public Type getType() {
return this.type;
}
@@ -119,25 +105,33 @@ public class FunctionRegistration<T> implements BeanNameAware {
}
public FunctionRegistration<T> type(Type type) {
return type(FunctionType.of(type));
}
public FunctionRegistration<T> type(FunctionType type) {
Type t = FunctionTypeUtils.discoverFunctionTypeFromClass(this.target.getClass());
if (t == null) { // only valid for Kafka Stream KStream[] return type.
Type discoveredFunctionType = FunctionTypeUtils.discoverFunctionTypeFromClass(this.target.getClass());
if (discoveredFunctionType == null) { // only valid for Kafka Stream KStream[] return type.
return null;
}
FunctionType discoveredFunctionType = FunctionType.of(t);
Class<?> inputType = TypeResolver.resolveRawClass(discoveredFunctionType.getInputType(), null);
Class<?> outputType = TypeResolver.resolveRawClass(discoveredFunctionType.getOutputType(), null);
if (!(inputType.isAssignableFrom(TypeResolver.resolveRawClass(type.getInputType(), null))
&& outputType.isAssignableFrom(TypeResolver.resolveRawClass(type.getOutputType(), null)))) {
throw new IllegalStateException("Discovered function type does not match provided function type. Discovered: "
+ discoveredFunctionType + "; Provided: " + type);
}
this.type = type;
Class<?> inputType = FunctionTypeUtils.getRawType(FunctionTypeUtils.getInputType(discoveredFunctionType));
Class<?> outputType = FunctionTypeUtils.getRawType(FunctionTypeUtils.getOutputType(discoveredFunctionType));
if (inputType != null && inputType != Object.class && outputType != null && outputType != Object.class) {
Assert.isTrue((inputType.isAssignableFrom(FunctionTypeUtils.getRawType(FunctionTypeUtils.getInputType(type)))
&& outputType.isAssignableFrom(FunctionTypeUtils.getRawType(FunctionTypeUtils.getOutputType(type)))),
"Discovered function type does not match provided function type. Discovered: "
+ discoveredFunctionType + "; Provided: " + type);
}
else if (inputType == null && outputType != Object.class) {
Assert.isTrue(outputType.isAssignableFrom(FunctionTypeUtils.getRawType(FunctionTypeUtils.getOutputType(type))),
"Discovered function type does not match provided function type. Discovered: "
+ discoveredFunctionType + "; Provided: " + type);
}
else if (outputType == null && inputType != Object.class) {
Assert.isTrue(inputType.isAssignableFrom(FunctionTypeUtils.getRawType(FunctionTypeUtils.getInputType(type))),
"Discovered function type does not match provided function type. Discovered: "
+ discoveredFunctionType + "; Provided: " + type);
}
return this;
}
@@ -175,52 +169,30 @@ public class FunctionRegistration<T> implements BeanNameAware {
*
*/
@SuppressWarnings({ "unchecked", "rawtypes" })
public <S> FunctionRegistration<S> wrap() {
this.isFunctionSignatureSupported();
FunctionRegistration<S> result;
if (this.type == null) {
result = (FunctionRegistration<S>) this;
}
else if (this.target instanceof RoutingFunction) {
S target = (S) this.target;
result = new FunctionRegistration<S>(target);
result.type(this.type.getType());
result = result.target(target).names(this.names)
.type(result.type.wrap(Flux.class)).properties(this.properties);
}
else {
S target = (S) this.target;
result = new FunctionRegistration<S>(target);
result.type(this.type.getType());
if (!this.type.isWrapper()) {
target = target instanceof Supplier
? (S) new FluxSupplier((Supplier<?>) target)
: target instanceof Function
? (S) new FluxFunction((Function<?, ?>) target)
: (S) new FluxConsumer((Consumer<?>) target);
}
else if (Mono.class.isAssignableFrom(this.type.getOutputWrapper())) {
target = target instanceof Supplier
? (S) new MonoSupplier((Supplier<?>) target)
: (S) new FluxToMonoFunction((Function<?, ?>) target);
}
else if (Mono.class.isAssignableFrom(this.type.getInputWrapper())) {
target = (S) new MonoToFluxFunction((Function) target);
}
else if (target instanceof Consumer) {
target = (S) new FluxedConsumer((Consumer<?>) target);
}
else if (target instanceof Function) {
target = (S) new FluxedFunction((Function<?, ?>) target);
}
result = result.target(target).names(this.names)
.type(result.type.wrap(Flux.class)).properties(this.properties);
}
return result;
}
// @SuppressWarnings({ "unchecked", "rawtypes" })
// public <S> FunctionRegistration<S> wrap() {
// this.isFunctionSignatureSupported();
// FunctionRegistration<S> result;
// if (this.type == null) {
// result = (FunctionRegistration<S>) this;
// }
// else if (this.target instanceof RoutingFunction) {
// S target = (S) this.target;
// result = new FunctionRegistration<S>(target);
// result.type(this.type.getType());
// result = result.target(target).names(this.names)
// .type(result.type.wrap(Flux.class)).properties(this.properties);
// }
// else {
// S target = (S) this.target;
// result = new FunctionRegistration<S>(target);
// result.type(this.type.getType());
// result = result.target(target).names(this.names)
// .type(result.type.wrap(Flux.class)).properties(this.properties);
// }
//
// return result;
// }
@Override
public void setBeanName(String name) {
@@ -229,12 +201,12 @@ public class FunctionRegistration<T> implements BeanNameAware {
}
}
private void isFunctionSignatureSupported() {
if (type != null) {
Assert.isTrue(!(Mono.class.isAssignableFrom(this.type.getOutputWrapper())
&& Mono.class.isAssignableFrom(this.type.getInputWrapper())),
"Function<Mono, Mono> is not supported.");
}
}
// private void isFunctionSignatureSupported() {
// if (type != null) {
// Assert.isTrue(!(Mono.class.isAssignableFrom(this.type.getOutputWrapper())
// && Mono.class.isAssignableFrom(this.type.getInputWrapper())),
// "Function<Mono, Mono> is not supported.");
// }
// }
}

549
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionType.java
View File

@@ -1,549 +0,0 @@
/*
* Copyright 2012-2019 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
*
* https://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.function.context;
import java.lang.reflect.Field;
import java.lang.reflect.Modifier;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import reactor.core.publisher.Flux;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.support.SpringFactoriesLoader;
import org.springframework.messaging.Message;
import org.springframework.util.ObjectUtils;
import org.springframework.util.ReflectionUtils;
/**
* @author Dave Syer
* @author Oleg Zhurakousky
*
*/
public class FunctionType {
/**
* Unclassified function types.
*/
public static FunctionType UNCLASSIFIED = new FunctionType(ResolvableType
.forClassWithGenerics(Function.class, Object.class, Object.class).getType());
private static List<WrapperDetector> transformers;
private Type type;
private Class<?> inputType;
private Class<?> outputType;
private Class<?> inputWrapper;
private Class<?> outputWrapper;
private boolean message;
public FunctionType(Type type) {
this.type = functionType(type);
this.inputWrapper = findType(ParamType.INPUT_WRAPPER);
this.outputWrapper = findType(ParamType.OUTPUT_WRAPPER);
this.inputType = findType(ParamType.INPUT);
this.outputType = findType(ParamType.OUTPUT);
this.message = messageType();
resetType();
}
/*
* Experimental for now. Used (reflectively) in FunctionCreatorConfiguration to effectively
* map an existing FunctionType created by one class loader to another.
*/
@SuppressWarnings("unused") // it is used
private FunctionType(Object functionType) throws Exception {
Field[] fields = functionType.getClass().getDeclaredFields();
for (Field field : fields) {
if (!Modifier.isStatic(field.getModifiers())) {
field.setAccessible(true);
Field thisField = ReflectionUtils.findField(this.getClass(), field.getName());
thisField.setAccessible(true);
thisField.set(this, field.get(functionType));
}
}
}
public static boolean isWrapper(Type type) {
if (type instanceof ParameterizedType) {
type = ((ParameterizedType) type).getRawType();
}
if (transformers == null) {
transformers = new ArrayList<>();
transformers.addAll(
SpringFactoriesLoader.loadFactories(WrapperDetector.class, null));
}
for (WrapperDetector transformer : transformers) {
if (transformer.isWrapper(type)) {
return true;
}
}
return false;
}
public static FunctionType of(Type function) {
FunctionType ft = new FunctionType(function);
if (!ft.isWrapper() && !(ft.type instanceof ParameterizedType)) {
Type[] genericInterfaces = ((Class<?>) function).getGenericInterfaces();
if (!ObjectUtils.isEmpty(genericInterfaces)) {
ft.type = genericInterfaces[0];
}
}
return ft;
}
public static FunctionType from(Class<?> input) {
return new FunctionType(ResolvableType
.forClassWithGenerics(Function.class, input, Object.class).getType());
}
public static FunctionType supplier(Class<?> input) {
return new FunctionType(
ResolvableType.forClassWithGenerics(Supplier.class, input).getType());
}
public static FunctionType consumer(Class<?> input) {
return new FunctionType(
ResolvableType.forClassWithGenerics(Consumer.class, input).getType());
}
public static FunctionType compose(FunctionType input, FunctionType output) {
ResolvableType inputGeneric = input(input);
ResolvableType outputGeneric = output(output);
if (!isWrapper(outputGeneric.getType())) {
ResolvableType inputOutput = output(input);
if (isWrapper(inputOutput.getType())) {
outputGeneric = wrap(input,
extractClass(inputOutput.getType(), ParamType.OUTPUT_WRAPPER),
extractClass(outputGeneric.getType(), ParamType.OUTPUT));
}
}
return new FunctionType(ResolvableType
.forClassWithGenerics(Function.class, inputGeneric, outputGeneric)
.getType());
}
public Type getType() {
return this.type;
}
public Class<?> getInputWrapper() {
return this.inputWrapper;
}
public Class<?> getOutputWrapper() {
return this.outputWrapper;
}
public Class<?> getInputType() {
return this.inputType;
}
public Class<?> getOutputType() {
return this.outputType;
}
public boolean isMessage() {
return this.message;
}
public boolean isWrapper() {
return isWrapper(getInputWrapper()) || isWrapper(getOutputWrapper());
}
public FunctionType to(Class<?> output) {
ResolvableType inputGeneric = input(this);
ResolvableType outputGeneric = output(output);
return new FunctionType(ResolvableType
.forClassWithGenerics(Function.class, inputGeneric, outputGeneric)
.getType());
}
public FunctionType message() {
if (isMessage()) {
return this;
}
ResolvableType inputGeneric = message(getInputType());
ResolvableType outputGeneric = message(getOutputType());
if (isWrapper(getInputWrapper())) {
inputGeneric = ResolvableType.forClassWithGenerics(getInputWrapper(),
inputGeneric);
outputGeneric = ResolvableType.forClassWithGenerics(getInputWrapper(),
outputGeneric);
}
return new FunctionType(ResolvableType
.forClassWithGenerics(Function.class, inputGeneric, outputGeneric)
.getType());
}
public FunctionType wrap(Class<?> input, Class<?> output) {
if (!isWrapper(input) && !isWrapper(output)) {
return this;
}
else if (isWrapper(input) && isWrapper(output)) {
if (input.isAssignableFrom(getInputWrapper())
&& output.isAssignableFrom(getOutputWrapper())) {
return this;
}
return new FunctionType(ResolvableType.forClassWithGenerics(Function.class,
wrapper(input, getInputType()), wrapper(output, getOutputType()))
.getType());
}
else {
throw new IllegalArgumentException("Both wrapper types must be wrappers in ("
+ input + ", " + output + ")");
}
}
public FunctionType wrap(Class<?> wrapper) {
return wrap(wrapper, wrapper);
}
@Override
public int hashCode() {
final int prime = 31;
int result = 1;
result = prime * result
+ ((this.inputType == null) ? 0 : this.inputType.toString().hashCode());
result = prime * result + ((this.inputWrapper == null) ? 0
: this.inputWrapper.toString().hashCode());
result = prime * result + (this.message ? 1231 : 1237);
result = prime * result
+ ((this.outputType == null) ? 0 : this.outputType.toString().hashCode());
result = prime * result + ((this.outputWrapper == null) ? 0
: this.outputWrapper.toString().hashCode());
return result;
}
public String toString() {
if (this.inputType == Void.class) {
return this.type.toString() + ", which is effectively a Supplier<"
+ this.outputType + ">";
}
else if (this.outputType == Void.class) {
return this.type.toString() + ", which is effectively a Consumer<"
+ this.inputType + ">";
}
return this.type.toString();
}
@Override
public boolean equals(Object obj) {
if (this == obj) {
return true;
}
if (obj == null) {
return false;
}
if (getClass() != obj.getClass()) {
return false;
}
FunctionType other = (FunctionType) obj;
if (this.inputType == null) {
if (other.inputType != null) {
return false;
}
}
else if (!this.inputType.toString().equals(other.inputType.toString())) {
return false;
}
if (this.inputWrapper == null) {
if (other.inputWrapper != null) {
return false;
}
}
else if (!this.inputWrapper.toString().equals(other.inputWrapper.toString())) {
return false;
}
if (this.message != other.message) {
return false;
}
if (this.outputType == null) {
if (other.outputType != null) {
return false;
}
}
else if (!this.outputType.toString().equals(other.outputType.toString())) {
return false;
}
if (this.outputWrapper == null) {
if (other.outputWrapper != null) {
return false;
}
}
else if (!this.outputWrapper.toString().equals(other.outputWrapper.toString())) {
return false;
}
return true;
}
private static ResolvableType wrap(FunctionType input, Class<?> wrapper,
Class<?> type) {
return input.isMessage() ? wrap(wrapper, message(type))
: ResolvableType.forClassWithGenerics(wrapper, type);
}
private static ResolvableType wrap(Class<?> wrapper, ResolvableType type) {
return ResolvableType.forClassWithGenerics(wrapper, type);
}
private static ResolvableType message(Class<?> type) {
return ResolvableType.forClassWithGenerics(Message.class, type);
}
private static ResolvableType input(FunctionType type) {
return type.input(type.getInputType());
}
private static ResolvableType output(FunctionType type) {
return type.output(type.getOutputType());
}
private static Class<?> extractClass(Type param, ParamType paramType) {
if (param instanceof ParameterizedType) {
ParameterizedType concrete = (ParameterizedType) param;
param = concrete.getRawType();
}
if (param == null) {
// Last ditch attempt to guess: Flux<String>
if (paramType.isWrapper()) {
param = Flux.class;
}
else {
param = String.class;
}
}
Class<?> result = param instanceof Class ? (Class<?>) param : null;
// TODO: cache result
return result;
}
private ResolvableType wrapper(Class<?> wrapper, Class<?> type) {
return wrap(this, wrapper, type);
}
private ResolvableType output(Class<?> type) {
ResolvableType generic;
ResolvableType raw = ResolvableType.forClass(type);
if (isMessage()) {
raw = ResolvableType.forClassWithGenerics(Message.class, raw);
}
if (FunctionType.isWrapper(getOutputWrapper())) {
generic = ResolvableType.forClassWithGenerics(getOutputWrapper(), raw);
}
else {
generic = raw;
}
return generic;
}
private ResolvableType input(Class<?> type) {
ResolvableType generic;
ResolvableType raw = ResolvableType.forClass(type);
if (isMessage()) {
raw = ResolvableType.forClassWithGenerics(Message.class, raw);
}
if (FunctionType.isWrapper(getInputWrapper())) {
generic = ResolvableType.forClassWithGenerics(getInputWrapper(), raw);
}
else {
generic = raw;
}
return generic;
}
private Class<?> findType(ParamType paramType) {
int index = paramType.isOutput() ? 1 : 0;
Type type = this.type;
if (Supplier.class.isAssignableFrom(extractClass(this.type, null))) {
if (paramType.isInput()) {
return Void.class;
}
}
boolean found = false;
while (!found && type instanceof Class && type != Object.class) {
Class<?> clz = (Class<?>) type;
for (Type iface : clz.getGenericInterfaces()) {
if (iface.getTypeName().startsWith("java.util.function")) {
type = iface;
found = true;
break;
}
}
if (!found) {
type = clz.getSuperclass();
}
}
Type param = extractType(type, paramType, index);
if (param != null) {
Class<?> result = extractClass(param, paramType);
if (result != null) {
return result;
}
}
return Object.class;
}
private void resetType() {
if (!this.type.getTypeName().contains("EnhancerBySpringCGLIB")) {
return;
}
Type type = this.type;
boolean found = false;
while (!found && type instanceof Class && type != Object.class) {
Class<?> clz = (Class<?>) type;
for (Type iface : clz.getGenericInterfaces()) {
if (iface.getTypeName().startsWith("java.util.function")) {
type = iface;
found = true;
break;
}
}
if (!found) {
type = clz.getSuperclass();
}
}
this.type = type;
}
private Type extractType(Type type, ParamType paramType, int index) {
Type param;
if (type instanceof ParameterizedType) {
ParameterizedType parameterizedType = (ParameterizedType) type;
if (parameterizedType.getActualTypeArguments().length == 1) {
if (isVoid(parameterizedType, paramType)) {
return Void.class;
}
// There's only one
index = 0;
}
Type typeArgumentAtIndex = parameterizedType.getActualTypeArguments()[index];
if (typeArgumentAtIndex instanceof ParameterizedType
&& !paramType.isWrapper()) {
if (FunctionType.isWrapper(
((ParameterizedType) typeArgumentAtIndex).getRawType())) {
param = ((ParameterizedType) typeArgumentAtIndex)
.getActualTypeArguments()[0];
param = extractNestedType(paramType, param);
}
else {
param = extractNestedType(paramType, typeArgumentAtIndex);
}
}
else {
param = extractNestedType(paramType, typeArgumentAtIndex);
}
}
else {
if (type != null) {
Type[] interfaces = ((Class<?>) type).getGenericInterfaces();
for (Type ifc : interfaces) {
Type value = extractType(ifc, paramType, index);
if (value != Object.class) {
return value;
}
}
}
param = Object.class;
}
return param;
}
private boolean isVoid(ParameterizedType parameterizedType, ParamType paramType) {
Class<?> rawType = extractClass(parameterizedType.getRawType(), paramType);
if (Consumer.class.isAssignableFrom(rawType) && paramType.isOutput()) {
return true;
}
if (Supplier.class.isAssignableFrom(rawType) && paramType.isInput()) {
return true;
}
return false;
}
private Type extractNestedType(ParamType paramType, Type param) {
if (!paramType.isInnerWrapper() && param instanceof ParameterizedType) {
if (((ParameterizedType) param).getRawType().getTypeName()
.startsWith(Message.class.getName())) {
param = ((ParameterizedType) param).getActualTypeArguments()[0];
}
}
return param;
}
private Type functionType(Type type) {
if (Supplier.class.isAssignableFrom(extractClass(type, ParamType.OUTPUT))) {
Type product = extractType(type, ParamType.OUTPUT, 0);
Class<?> output = extractClass(product, ParamType.OUTPUT);
if (output != null) {
if (FunctionRegistration.class.isAssignableFrom(output)) {
type = extractType(product, ParamType.OUTPUT, 0);
}
else if (Function.class.isAssignableFrom(output)
|| Supplier.class.isAssignableFrom(output)
|| Consumer.class.isAssignableFrom(output)) {
type = product;
}
}
}
return type;
}
private boolean messageType() {
Class<?> inputType = findType(ParamType.INPUT_INNER_WRAPPER);
Class<?> outputType = findType(ParamType.OUTPUT_INNER_WRAPPER);
return inputType.getName().startsWith(Message.class.getName())
|| Message.class.isAssignableFrom(inputType)
|| outputType.getName().startsWith(Message.class.getName())
|| Message.class.isAssignableFrom(outputType);
}
enum ParamType {
INPUT, OUTPUT, INPUT_WRAPPER, OUTPUT_WRAPPER, INPUT_INNER_WRAPPER, OUTPUT_INNER_WRAPPER;
public boolean isOutput() {
return this == OUTPUT || this == OUTPUT_WRAPPER
|| this == OUTPUT_INNER_WRAPPER;
}
public boolean isInput() {
return this == INPUT || this == INPUT_WRAPPER || this == INPUT_INNER_WRAPPER;
}
public boolean isWrapper() {
return this == OUTPUT_WRAPPER || this == INPUT_WRAPPER;
}
public boolean isInnerWrapper() {
return this == OUTPUT_INNER_WRAPPER || this == INPUT_INNER_WRAPPER;
}
}
}

2
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionalSpringApplication.java
View File

@@ -153,7 +153,7 @@ public class FunctionalSpringApplication
context.registerBean("function", FunctionRegistration.class,
() -> new FunctionRegistration<>(
handler(context, function, functionType))
.type(FunctionType.of(functionType)));
.type(functionType));
}
private Object handler(GenericApplicationContext generic, Object handler,

439
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/AbstractComposableFunctionRegistry.java
View File

@@ -1,439 +0,0 @@
/*
* Copyright 2019-2019 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
*
* https://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.function.context.catalog;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import reactor.core.publisher.Flux;
import org.springframework.cloud.function.context.FunctionRegistration;
import org.springframework.cloud.function.context.FunctionRegistry;
import org.springframework.cloud.function.context.FunctionType;
import org.springframework.cloud.function.context.config.RoutingFunction;
import org.springframework.cloud.function.core.FluxToMonoFunction;
import org.springframework.cloud.function.core.IsolatedConsumer;
import org.springframework.cloud.function.core.IsolatedFunction;
import org.springframework.cloud.function.core.IsolatedSupplier;
import org.springframework.cloud.function.core.MonoToFluxFunction;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.context.EnvironmentAware;
import org.springframework.core.env.Environment;
import org.springframework.core.env.StandardEnvironment;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
/**
* Base implementation of {@link FunctionRegistry} which supports function composition
* during lookups. For example if this registry contains function 'a' and 'b' you can
* compose them into a single function by simply piping two names together during the
* lookup {@code this.lookup(Function.class, "a|b")}.
*
* Comma ',' is also supported as composition delimiter (e.g., {@code "a,b"}).
*
* @author Oleg Zhurakousky
* @author Dave Syer
* @since 2.1
*
*/
public abstract class AbstractComposableFunctionRegistry implements FunctionRegistry,
ApplicationEventPublisherAware, EnvironmentAware {
private final Map<String, Object> functions = new ConcurrentHashMap<>();
private final Map<Object, String> names = new ConcurrentHashMap<>();
private final Map<String, FunctionType> types = new ConcurrentHashMap<>();
private Environment environment = new StandardEnvironment();
protected ApplicationEventPublisher applicationEventPublisher;
@SuppressWarnings("unchecked")
@Override
public <T> T lookup(Class<?> type, String name) {
String functionDefinitionName = !StringUtils.hasText(name)
&& this.environment.containsProperty("spring.cloud.function.definition")
? this.environment.getProperty("spring.cloud.function.definition")
: name;
return (T) this.doLookup(type, functionDefinitionName);
}
@SuppressWarnings("serial")
@Override
public Set<String> getNames(Class<?> type) {
if (type == null) {
return new HashSet<String>(getSupplierNames()) {
{
addAll(getFunctionNames());
}
};
}
if (Supplier.class.isAssignableFrom(type)) {
return this.getSupplierNames();
}
if (Function.class.isAssignableFrom(type)) {
return this.getFunctionNames();
}
return Collections.emptySet();
}
/**
* Returns the names of available Suppliers.
* @return immutable {@link Set} of available {@link Supplier} names.
*/
public Set<String> getSupplierNames() {
return this.functions.entrySet().stream()
.filter(entry -> entry.getValue() instanceof Supplier)
.map(entry -> entry.getKey())
.collect(Collectors.toSet());
}
/**
* Returns the names of available Functions.
* @return immutable {@link Set} of available {@link Function} names.
*/
public Set<String> getFunctionNames() {
return this.functions.entrySet().stream()
.filter(entry -> !(entry.getValue() instanceof Supplier))
.map(entry -> entry.getKey())
.collect(Collectors.toSet());
}
public boolean hasSuppliers() {
return !CollectionUtils.isEmpty(getSupplierNames());
}
public boolean hasFunctions() {
return !CollectionUtils.isEmpty(getFunctionNames());
}
/**
* The size of this catalog, which is the count of all Suppliers,
* Function and Consumers currently registered.
*
* @return the count of all Suppliers, Function and Consumers currently registered.
*/
@Override
public int size() {
return this.functions.size();
}
public FunctionType getFunctionType(String name) {
return this.types.get(name);
}
/**
* A reverse lookup where one can determine the actual name of the function reference.
* @param function should be an instance of {@link Supplier}, {@link Function} or
* {@link Consumer};
* @return the name of the function or null.
*/
public String lookupFunctionName(Object function) {
return this.names.containsKey(function) ? this.names.get(function) : null;
}
@Override
public void setApplicationEventPublisher(
ApplicationEventPublisher applicationEventPublisher) {
this.applicationEventPublisher = applicationEventPublisher;
}
@Override
public void setEnvironment(Environment environment) {
this.environment = environment;
}
public FunctionRegistration<?> getRegistration(Object function) {
String functionName = function == null ? null
: this.lookupFunctionName(function);
if (StringUtils.hasText(functionName)) {
FunctionRegistration<?> registration = new FunctionRegistration<Object>(
function, functionName);
FunctionType functionType = this.findType(registration, functionName);
return registration.type(functionType.getType());
}
return null;
}
@Override
public <T> void register(FunctionRegistration<T> functionRegistration) {
Assert.notEmpty(functionRegistration.getNames(),
"'registration' must contain at least one name before it is registered in catalog.");
register(functionRegistration, functionRegistration.getNames().iterator().next());
}
/**
* Registers function wrapped by the provided FunctionRegistration with
* this FunctionRegistry.
*
* @param registration instance of {@link FunctionRegistration}
* @param key the name of the function
*/
protected void register(FunctionRegistration<?> registration, String key) {
Object target = registration.getTarget();
if (registration.getType() != null) {
this.addType(key, registration.getType());
}
else {
FunctionType functionType = findType(registration, key);
if (functionType == null) {
return; // TODO fixme
}
this.addType(key, functionType);
registration.type(functionType.getType());
}
Class<?> type;
registration = isolated(registration).wrap();
target = registration.getTarget();
if (target instanceof Supplier) {
type = Supplier.class;
for (String name : registration.getNames()) {
this.addSupplier(name, (Supplier<?>) registration.getTarget());
}
}
else if (target instanceof Function) {
type = Function.class;
for (String name : registration.getNames()) {
this.addFunction(name, (Function<?, ?>) registration.getTarget());
}
}
else {
return;
}
this.addName(registration.getTarget(), key);
if (this.applicationEventPublisher != null) {
this.applicationEventPublisher.publishEvent(new FunctionRegistrationEvent(
registration.getTarget(), type, registration.getNames()));
}
}
protected FunctionType findType(FunctionRegistration<?> functionRegistration, String name) {
return functionRegistration.getType() != null
? functionRegistration.getType()
: this.getFunctionType(name);
}
protected void addSupplier(String name, Supplier<?> supplier) {
this.functions.put(name, supplier);
}
protected void addFunction(String name, Function<?, ?> function) {
this.functions.put(name, function);
}
protected void addType(String name, FunctionType functionType) {
this.types.computeIfAbsent(name, str -> functionType);
}
protected void addName(Object function, String name) {
this.names.put(function, name);
}
@SuppressWarnings({ "rawtypes", "unchecked" })
private FunctionRegistration<?> isolated(FunctionRegistration<?> input) {
FunctionRegistration<Object> registration = (FunctionRegistration<Object>) input;
Object target = registration.getTarget();
boolean isolated = getClass().getClassLoader() != target.getClass()
.getClassLoader();
if (isolated) {
if (target instanceof Supplier<?> && isolated) {
target = new IsolatedSupplier((Supplier<?>) target);
}
else if (target instanceof Function<?, ?>) {
target = new IsolatedFunction((Function<?, ?>) target);
}
else if (target instanceof Consumer<?>) {
target = new IsolatedConsumer((Consumer<?>) target);
}
}
registration.target(target);
return registration;
}
private Object compose(String name, Map<String, Object> lookup) {
name = name.replaceAll(",", "|").trim();
Object composedFunction = null;
if (lookup.containsKey(name)) {
composedFunction = lookup.get(name);
}
else if (name.equals("") && lookup.size() >= 1 && lookup.size() <= 2) { // we may have RoutingFunction function
String functionName = lookup.keySet().stream()
.filter(fName -> !fName.equals(RoutingFunction.FUNCTION_NAME))
.findFirst().orElseGet(() -> null);
composedFunction = lookup.get(functionName);
}
else {
String[] stages = StringUtils.delimitedListToStringArray(name, "|");
AtomicBoolean supplierPresent = new AtomicBoolean();
List<FunctionRegistration<?>> composableFunctions = Stream.of(stages)
.map(funcName -> find(funcName, supplierPresent.get()))
.filter(x -> x != null)
.peek(f -> supplierPresent.set(f.getTarget() instanceof Supplier))
.collect(Collectors.toList());
FunctionRegistration<?> composedRegistration = composableFunctions
.stream().reduce((a, z) -> composeFunctions(a, z))
.orElseGet(() -> null);
if (composedRegistration != null
&& composedRegistration.getTarget() != null
&& !this.types.containsKey(name)) {
composedFunction = composedRegistration.getTarget();
this.addType(name, composedRegistration.getType());
this.addName(composedFunction, name);
if (composedFunction instanceof Function || composedFunction instanceof Consumer) {
this.addFunction(name, (Function<?, ?>) composedFunction);
}
else if (composedFunction instanceof Supplier) {
this.addSupplier(name, (Supplier<?>) composedFunction);
}
}
}
return composedFunction;
}
private FunctionRegistration<?> find(String name, boolean supplierFound) {
Object result = this.functions.get(name);
if (result == null && !StringUtils.hasText(name)) {
if (supplierFound && this.getFunctionNames().size() == 1) {
result = this.functions.get(this.getFunctionNames().iterator().next());
}
else if (!supplierFound && this.getSupplierNames().size() == 1) {
result = this.functions.get(this.getSupplierNames().iterator().next());
}
}
return getRegistration(result);
}
@SuppressWarnings("unchecked")
private FunctionRegistration<?> composeFunctions(FunctionRegistration<?> aReg,
FunctionRegistration<?> bReg) {
FunctionType aType = aReg.getType();
FunctionType bType = bReg.getType();
Object a = aReg.getTarget();
Object b = bReg.getTarget();
if (aType != null && bType != null) {
if (aType.isMessage() && !bType.isMessage()) {
bType = bType.message();
b = message(b);
}
}
Object composedFunction = null;
// if (a instanceof Supplier && b instanceof Function) {
// Supplier<Flux<Object>> supplier = (Supplier<Flux<Object>>) a;
// if (b instanceof FluxConsumer) {
// if (supplier instanceof FluxSupplier) {
// FluxConsumer<Object> fConsumer = ((FluxConsumer<Object>) b);
// composedFunction = (Supplier<Mono<Void>>) () -> Mono.from(
// supplier.get().compose(v -> fConsumer.apply(supplier.get())));
// }
// else {
// throw new IllegalStateException(
// "The provided supplier is finite (i.e., already composed with Consumer) "
// + "therefore it can not be composed with another consumer");
// }
// }
// else {
// Function<Object, Object> function = (Function<Object, Object>) b;
// composedFunction = (Supplier<Object>) () -> function
// .apply(supplier.get());
// }
// }
// else
if (a instanceof Function && b instanceof Function) {
Function<Object, Object> function1 = (Function<Object, Object>) a;
Function<Object, Object> function2 = (Function<Object, Object>) b;
if (function1 instanceof FluxToMonoFunction) {
if (function2 instanceof MonoToFluxFunction) {
composedFunction = function1.andThen(function2);
}
else {
throw new IllegalStateException(
"The provided function is finite (i.e., returns Mono<?>) "
+ "therefore it can *only* be composed with compatible function (i.e., Function<Mono, Flux>");
}
}
else if (function2 instanceof FluxToMonoFunction) {
composedFunction = new FluxToMonoFunction<Object, Object>(
((Function<Flux<Object>, Flux<Object>>) a).andThen(
((FluxToMonoFunction<Object, Object>) b).getTarget()));
}
else {
composedFunction = function1.andThen(function2);
}
}
else if (a instanceof Function && b instanceof Consumer) {
Function<Object, Object> function = (Function<Object, Object>) a;
Consumer<Object> consumer = (Consumer<Object>) b;
composedFunction = (Consumer<Object>) v -> consumer.accept(function.apply(v));
}
else {
throw new IllegalArgumentException(String
.format("Could not compose %s and %s", a.getClass(), b.getClass()));
}
String name = aReg.getNames().iterator().next() + "|"
+ bReg.getNames().iterator().next();
return new FunctionRegistration<>(composedFunction, name)
.type(FunctionType.compose(aType, bType));
}
private Object message(Object input) {
if (input instanceof Supplier) {
return new MessageSupplier((Supplier<?>) input);
}
if (input instanceof Consumer) {
return new MessageConsumer((Consumer<?>) input);
}
if (input instanceof Function) {
return new MessageFunction((Function<?, ?>) input);
}
return input;
}
private Object doLookup(Class<?> type, String name) {
Object function = this.compose(name, this.functions);
if (function != null && type != null && !type.isAssignableFrom(function.getClass())) {
function = null;
}
return function;
}
}

153
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionInspector.java
View File

@@ -1,153 +0,0 @@
/*
* Copyright 2012-2020 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
*
* https://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.function.context.catalog;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.lang.reflect.TypeVariable;
import java.lang.reflect.WildcardType;
import net.jodah.typetools.TypeResolver;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.cloud.function.context.FunctionRegistration;
import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
/**
* @author Dave Syer
* @author Oleg Zhurakousky
*
* @deprecated since 3.1 no longer used by the framework
*/
@Deprecated
public interface FunctionInspector {
FunctionRegistration<?> getRegistration(Object function);
/**
*
* @deprecated since 3.1 no longer used by the framework
*/
@Deprecated
default boolean isMessage(Object function) {
if (function == null) {
return false;
}
return ((FunctionInvocationWrapper) function).isInputTypeMessage();
}
/**
*
* @deprecated since 3.1 no longer used by the framework
*/
@Deprecated
default Class<?> getInputType(Object function) {
if (function == null) {
return Object.class;
}
Type type = ((FunctionInvocationWrapper) function).getInputType();
Class<?> inputType;
if (type instanceof ParameterizedType) {
if (function != null && (((FunctionInvocationWrapper) function).isInputTypePublisher() || ((FunctionInvocationWrapper) function).isInputTypeMessage())) {
inputType = TypeResolver.resolveRawClass(FunctionTypeUtils.getImmediateGenericType(type, 0), null);
}
else {
inputType = ((FunctionInvocationWrapper) function).getRawInputType();
}
}
else {
inputType = type instanceof TypeVariable || type instanceof WildcardType ? Object.class : (Class<?>) type;
}
return inputType;
}
/**
*
* @deprecated since 3.1 no longer used by the framework
*/
@Deprecated
default Class<?> getOutputType(Object function) {
if (function == null) {
return Object.class;
}
Type type = ((FunctionInvocationWrapper) function).getOutputType();
Class<?> outputType;
if (type instanceof ParameterizedType) {
if (function != null && ((FunctionInvocationWrapper) function).isOutputTypePublisher() || ((FunctionInvocationWrapper) function).isOutputTypeMessage()) {
outputType = TypeResolver.resolveRawClass(FunctionTypeUtils.getImmediateGenericType(type, 0), null);
}
else {
outputType = ((FunctionInvocationWrapper) function).getRawOutputType();
}
}
else {
outputType = type instanceof TypeVariable || type instanceof WildcardType ? Object.class : (Class<?>) type;
}
return outputType;
}
/**
*
* @deprecated since 3.1 no longer used by the framework
*/
@Deprecated
default Class<?> getInputWrapper(Object function) {
Class c = function == null ? Object.class : TypeResolver.resolveRawClass(((FunctionInvocationWrapper) function).getInputType(), null);
if (Flux.class.isAssignableFrom(c)) {
return c;
}
else if (Mono.class.isAssignableFrom(c)) {
return c;
}
else {
return this.getInputType(function);
}
}
/**
*
* @deprecated since 3.1 no longer used by the framework
*/
@Deprecated
default Class<?> getOutputWrapper(Object function) {
Class c = function == null ? Object.class : TypeResolver.resolveRawClass(((FunctionInvocationWrapper) function).getOutputType(), null);
if (Flux.class.isAssignableFrom(c)) {
return c;
}
else if (Mono.class.isAssignableFrom(c)) {
return c;
}
else {
return this.getOutputType(function);
}
}
/**
*
* @deprecated since 3.1 no longer used by the framework
*/
@Deprecated
default String getName(Object function) {
if (function == null) {
return null;
}
return ((FunctionInvocationWrapper) function).getFunctionDefinition();
}
}

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

@@ -71,12 +71,12 @@ class FunctionTypeConversionHelper {
this.conversionService = conversionService;
this.messageConverter = messageConverter;
this.functionRegistration = functionRegistration;
if ((this.functionRegistration.getType().getType()) instanceof ParameterizedType) {
this.functionArgumentTypes = ((ParameterizedType) this.functionRegistration.getType().getType())
if ((this.functionRegistration.getType()) instanceof ParameterizedType) {
this.functionArgumentTypes = ((ParameterizedType) this.functionRegistration.getType())
.getActualTypeArguments();
}
else {
this.functionArgumentTypes = new Type[] { this.functionRegistration.getType().getInputType() };
this.functionArgumentTypes = new Type[] { FunctionTypeUtils.getInputType(this.functionRegistration.getType()) };
}
}
@@ -227,7 +227,7 @@ class FunctionTypeConversionHelper {
}
}
else {
Assert.isTrue(!Publisher.class.isAssignableFrom(this.functionRegistration.getType().getInputWrapper()),
Assert.isTrue(!FunctionTypeUtils.isPublisher(FunctionTypeUtils.getInputType(this.functionRegistration.getType())),
"Invoking reactive function as imperative is not allowed. Function name(s): "
+ this.functionRegistration.getNames());
incoming = this.doConvertArgument(incoming, targetType, actualType);
@@ -244,7 +244,7 @@ class FunctionTypeConversionHelper {
: Flux.from((Publisher) incoming).map(value -> this.messageConverter.toMessage(value, headers));
}
else {
Assert.isTrue(!Publisher.class.isAssignableFrom(this.functionRegistration.getType().getInputWrapper()),
Assert.isTrue(!FunctionTypeUtils.isPublisher(FunctionTypeUtils.getInputType(this.functionRegistration.getType())),
"Invoking reactive function as imperative is not allowed. Function name(s): "
+ this.functionRegistration.getNames());
incoming = this.messageConverter.toMessage(incoming, headers);
@@ -296,7 +296,7 @@ class FunctionTypeConversionHelper {
incomingValue = incomingMessage;
}
else {
incomingValue = this.messageConverter.fromMessage((Message<?>) incomingMessage, targetType);
incomingValue = this.messageConverter.fromMessage(incomingMessage, targetType);
}
}
return incomingValue;

77
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeUtils.java
View File

@@ -41,7 +41,6 @@ import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.annotation.BeanFactoryAnnotationUtils;
import org.springframework.cloud.function.context.FunctionRegistration;
import org.springframework.cloud.function.context.FunctionType;
import org.springframework.cloud.function.context.config.FunctionContextUtils;
import org.springframework.cloud.function.context.config.RoutingFunction;
import org.springframework.context.support.GenericApplicationContext;
@@ -68,6 +67,21 @@ public final class FunctionTypeUtils {
}
public static Type functionType(Type input, Type output) {
return ResolvableType.forClassWithGenerics(Function.class,
ResolvableType.forType(input), ResolvableType.forType(output)).getType();
}
public static Type consumerType(Type input) {
return ResolvableType.forClassWithGenerics(Consumer.class,
ResolvableType.forType(input)).getType();
}
public static Type supplierType(Type output) {
return ResolvableType.forClassWithGenerics(Supplier.class,
ResolvableType.forType(output)).getType();
}
/**
* Will return 'true' if the provided type is a {@link Collection} type.
* This also includes collections wrapped in {@link Message}. For example,
@@ -176,6 +190,37 @@ public final class FunctionTypeUtils {
return null;
}
/**
* Discovers the function {@link Type} based on the signature of a factory method.
* For example, given the following method {@code Function<Message<Person>, Message<String>> uppercase()} of
* class Foo - {@code Type type = discoverFunctionTypeFromFunctionFactoryMethod(Foo.class, "uppercase");}
*
* @param clazz instance of Class containing the factory method
* @param methodName factory method name
* @return type of the function
*/
public static Type discoverFunctionTypeFromFunctionFactoryMethod(Class<?> clazz, String methodName) {
return discoverFunctionTypeFromFunctionFactoryMethod(ReflectionUtils.findMethod(clazz, methodName));
}
/**
* Discovers the function {@link Type} based on the signature of a factory method.
* For example, given the following method {@code Function<Message<Person>, Message<String>> uppercase()} of
* class Foo - {@code Type type = discoverFunctionTypeFromFunctionFactoryMethod(Foo.class, "uppercase");}
*
* @param method factory method
* @return type of the function
*/
public static Type discoverFunctionTypeFromFunctionFactoryMethod(Method method) {
return method.getGenericReturnType();
}
/**
* Unlike {@link #discoverFunctionTypeFromFunctionFactoryMethod(Class, String)}, this method discovers function
* type from the well known method of Function(apply), Supplier(get) or Consumer(accept).
* @param functionMethod functional method
* @return type of the function
*/
public static Type discoverFunctionTypeFromFunctionMethod(Method functionMethod) {
Assert.isTrue(
functionMethod.getName().equals("apply") ||
@@ -223,6 +268,26 @@ public final class FunctionTypeUtils {
return outputCount;
}
/**
* In the event the input type is {@link ParameterizedType} this method returns its generic type.
* @param functionType instance of function type
* @return generic type or input type
*/
public static Type getComponentTypeOfInputType(Type functionType) {
Type inputType = getInputType(functionType);
return getImmediateGenericType(inputType, 0);
}
/**
* In the event the output type is {@link ParameterizedType} this method returns its generic type.
* @param functionType instance of function type
* @return generic type or output type
*/
public static Type getComponentTypeOfOutputType(Type functionType) {
Type inputType = getOutputType(functionType);
return getImmediateGenericType(inputType, 0);
}
/**
* Returns input type of function type that represents Function or Consumer.
* @param functionType the Type of Function or Consumer
@@ -253,15 +318,15 @@ public final class FunctionTypeUtils {
@SuppressWarnings("rawtypes")
public static Type discoverFunctionType(Object function, String functionName, GenericApplicationContext applicationContext) {
if (function instanceof RoutingFunction) {
return FunctionType.of(FunctionContextUtils.findType(applicationContext.getBeanFactory(), functionName)).getType();
return FunctionContextUtils.findType(applicationContext.getBeanFactory(), functionName);
}
else if (function instanceof FunctionRegistration) {
return ((FunctionRegistration) function).getType().getType();
return ((FunctionRegistration) function).getType();
}
if (applicationContext.containsBean(functionName + FunctionRegistration.REGISTRATION_NAME_SUFFIX)) { // for Kotlin primarily
FunctionRegistration fr = applicationContext
.getBean(functionName + FunctionRegistration.REGISTRATION_NAME_SUFFIX, FunctionRegistration.class);
return fr.getType().getType();
return fr.getType();
}
boolean beanDefinitionExists = false;
@@ -277,13 +342,13 @@ public final class FunctionTypeUtils {
if (beanDefinitionExists) {
Type t = FunctionTypeUtils.getImmediateGenericType(type, 0);
if (t == null || t == Object.class) {
type = FunctionType.of(FunctionContextUtils.findType(applicationContext.getBeanFactory(), functionBeanDefinitionName)).getType();
type = FunctionContextUtils.findType(applicationContext.getBeanFactory(), functionBeanDefinitionName);
}
}
else if (!(type instanceof ParameterizedType)) {
String beanDefinitionName = discoverBeanDefinitionNameByQualifier(applicationContext.getBeanFactory(), functionName);
if (StringUtils.hasText(beanDefinitionName)) {
type = FunctionType.of(FunctionContextUtils.findType(applicationContext.getBeanFactory(), beanDefinitionName)).getType();
type = FunctionContextUtils.findType(applicationContext.getBeanFactory(), beanDefinitionName);
}
}
return type;

106
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/MessageFunction.java
View File

@@ -1,106 +0,0 @@
/*
* Copyright 2019-2019 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
*
* https://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.function.context.catalog;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.cloud.function.core.FluxConsumer;
import org.springframework.cloud.function.core.FluxFunction;
import org.springframework.cloud.function.core.FluxToMonoFunction;
import org.springframework.cloud.function.core.FluxedFunction;
import org.springframework.cloud.function.core.MonoToFluxFunction;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
/**
* @author Dave Syer
* @since 2.1
*/
public class MessageFunction
implements Function<Publisher<?>, Publisher<Message<?>>> {
private final Function<?, ?> delegate;
public MessageFunction(Function<?, ?> delegate) {
this.delegate = delegate;
}
@SuppressWarnings({ "unchecked", "rawtypes" })
@Override
public Publisher<Message<?>> apply(Publisher<?> input) {
Flux<Object> incomingFlux = Flux.from(input);
Flux<Message<?>> flux = incomingFlux.map(value -> {
if (!(value instanceof Message)) {
return MessageBuilder.withPayload(value).build();
}
return (Message<?>) value;
});
if (this.delegate instanceof FluxFunction) {
Function<Object, Object> target = (Function<Object, Object>) ((FluxFunction<?, ?>) this.delegate)
.getTarget();
return flux.map(
value -> MessageBuilder.withPayload(target.apply(value.getPayload()))
.copyHeaders(value.getHeaders()).build());
}
if (this.delegate instanceof MonoToFluxFunction) {
Function<Mono<Object>, Flux<Object>> target = ((MonoToFluxFunction<Object, Object>) this.delegate)
.getTarget();
return flux.next()
.flatMapMany(value -> target.apply(Mono.just(value.getPayload()))
.map(object -> MessageBuilder.withPayload(object)
.copyHeaders(value.getHeaders()).build()));
}
if (this.delegate instanceof FluxToMonoFunction) {
Function<Flux<Object>, Mono<Object>> target = ((FluxToMonoFunction<Object, Object>) this.delegate)
.getTarget();
AtomicReference<MessageHeaders> headers = new AtomicReference<>();
return target.apply(flux.map(messsage -> {
headers.set(messsage.getHeaders());
return messsage.getPayload();
})).map(payload -> MessageBuilder.withPayload(payload)
.copyHeaders(headers.get()).build());
}
if (this.delegate instanceof FluxConsumer) {
FluxConsumer<Object> target = ((FluxConsumer<Object>) this.delegate);
AtomicReference<MessageHeaders> headers = new AtomicReference<>();
Mono<Void> mapped = target.apply(flux.map(messsage -> {
headers.set(messsage.getHeaders());
return messsage.getPayload();
}));
return mapped.map(value -> MessageBuilder.createMessage(null, headers.get()));
}
// TODO: cover the case that delegate is actually Function<Flux,Flux>
if (this.delegate instanceof FluxedFunction) {
Function<Flux<Object>, Flux<Object>> target = ((FluxedFunction) this.delegate);
return (Flux) flux.map(value -> ((Message) value).getPayload()).transform(target);
}
Function function = this.delegate;
return flux.map(
value -> {
return MessageBuilder.withPayload(function.apply(value.getPayload()))
.copyHeaders(value.getHeaders()).build();
});
}
}

63
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/MessageSupplier.java
View File

@@ -1,63 +0,0 @@
/*
* Copyright 2019-2019 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
*
* https://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.function.context.catalog;
import java.util.function.Supplier;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.cloud.function.core.FluxSupplier;
import org.springframework.cloud.function.core.MonoSupplier;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
/**
* @author Dave Syer
*/
public class MessageSupplier implements Supplier<Publisher<Message<?>>> {
private Supplier<?> delegate;
public MessageSupplier(Supplier<?> delegate) {
this.delegate = delegate;
}
@Override
public Publisher<Message<?>> get() {
if (this.delegate instanceof FluxSupplier) {
return ((Flux<?>) this.delegate.get())
.map(value -> MessageBuilder.withPayload(value).build());
}
if (this.delegate instanceof MonoSupplier) {
return ((Mono<?>) this.delegate.get())
.map(value -> MessageBuilder.withPayload(value).build());
}
Object product = this.delegate.get();
if (product instanceof Publisher) {
return Flux.from((Publisher<?>) product)
.map(value -> MessageBuilder.withPayload(value).build());
}
if (product instanceof Iterable) {
return Flux.fromIterable((Iterable<?>) product)
.map(value -> MessageBuilder.withPayload(value).build());
}
return Mono.just(MessageBuilder.withPayload(product).build());
}
}

11
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java
View File

@@ -82,7 +82,7 @@ import org.springframework.util.StringUtils;
* @author Oleg Zhurakousky
*
*/
public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspector {
public class SimpleFunctionRegistry implements FunctionRegistry {
protected Log logger = LogFactory.getLog(this.getClass());
/*
* - do we care about FunctionRegistration after it's been registered? What additional value does it bring?
@@ -133,13 +133,6 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
}
}
@Override
public FunctionRegistration<?> getRegistration(Object function) {
throw new UnsupportedOperationException("FunctionInspector is deprecated. There is no need "
+ "to access FunctionRegistration directly since you can interogate the actual "
+ "looked-up function (see FunctionInvocationWrapper.");
}
public SimpleFunctionRegistry(ConversionService conversionService, CompositeMessageConverter messageConverter, JsonMapper jsonMapper) {
this(conversionService, messageConverter, jsonMapper, null, null);
}
@@ -269,7 +262,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
.findFirst()
.orElseGet(() -> null);
FunctionInvocationWrapper function = functionRegistration != null
? this.invocationWrapperInstance(functionName, functionRegistration.getTarget(), functionRegistration.getType().getType())
? this.invocationWrapperInstance(functionName, functionRegistration.getTarget(), functionRegistration.getType())
: null;
if (functionRegistration != null && functionRegistration.getProperties().containsKey("singleton")) {
try {

15
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/FunctionContextUtils.java
View File

@@ -29,7 +29,6 @@ import org.springframework.beans.factory.config.ConstructorArgumentValues;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.cloud.function.context.catalog.FunctionTypeUtils;
import org.springframework.cloud.function.core.FunctionFactoryMetadata;
import org.springframework.context.annotation.ScannedGenericBeanDefinition;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.Resource;
@@ -88,20 +87,6 @@ public abstract class FunctionContextUtils {
if (type != null) {
param = type.getType();
}
else {
Class<?> beanClass = definition.hasBeanClass() ? definition.getBeanClass() : null;
if (beanClass != null
&& !FunctionFactoryMetadata.class.isAssignableFrom(beanClass)) {
param = beanClass;
}
else {
Object bean = registry.getBean(actualName);
// could be FunctionFactoryMetadata. . . TODO investigate and fix
if (bean instanceof FunctionFactoryMetadata) {
param = ((FunctionFactoryMetadata<?>) bean).getFactoryMethod().getGenericReturnType();
}
}
}
}
return param;
}

1
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/converter/avro/AvroSchemaServiceManager.java
View File

@@ -63,7 +63,6 @@ public interface AvroSchemaServiceManager {
* @param writerSchema {@link Schema} writerSchema provided at run time
* @return datum reader which can be used to read Avro payload
*/
@SuppressWarnings({ "unchecked", "rawtypes" })
DatumReader<Object> getDatumReader(Class<? extends Object> type, Schema schema, Schema writerSchema);
/**

1
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/converter/avro/AvroSchemaServiceManagerImpl.java
View File

@@ -71,6 +71,7 @@ public class AvroSchemaServiceManagerImpl implements AvroSchemaServiceManager {
* @param schema {@link Schema} of object which needs to be serialized
* @return datum writer which can be used to write Avro payload
*/
@SuppressWarnings({ "unchecked", "rawtypes" })
@Override
public DatumWriter<Object> getDatumWriter(Class<?> type, Schema schema) {
DatumWriter<Object> writer;

88
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/message/MessageUtils.java
View File

@@ -16,18 +16,6 @@
package org.springframework.cloud.function.context.message;
import java.lang.reflect.Method;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import org.springframework.cloud.function.core.FluxWrapper;
import org.springframework.cloud.function.core.Isolated;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.ClassUtils;
import org.springframework.util.ReflectionUtils;
/**
* @author Dave Syer
* @author Oleg Zhurakousky
@@ -48,80 +36,4 @@ public abstract class MessageUtils {
* Value for 'target-protocol' typically use as header key.
*/
public static String SOURCE_TYPE = "source-type";
/**
* Create a message for the handler. If the handler is a wrapper for a function in an
* isolated class loader, then the message will be created with the target class
* loader (therefore the {@link Message} class must be on the classpath of the target
* class loader).
* @param handler the function that will be applied to the message
* @param payload the payload of the message
* @param headers the headers for the message
* @return a message with the correct class loader
*/
public static Object create(Object handler, Object payload,
Map<String, Object> headers) {
if (handler instanceof FluxWrapper) {
handler = ((FluxWrapper<?>) handler).getTarget();
}
if (payload instanceof Message) {
headers = new HashMap<>(headers);
headers.putAll(((Message<?>) payload).getHeaders());
payload = ((Message<?>) payload).getPayload();
}
if (!(handler instanceof Isolated)) {
return MessageBuilder.withPayload(payload).copyHeaders(headers).build();
}
ClassLoader classLoader = ((Isolated) handler).getClassLoader();
Class<?> builder = ClassUtils.resolveClassName(MessageBuilder.class.getName(),
classLoader);
Method withPayload = ClassUtils.getMethod(builder, "withPayload", Object.class);
Method copyHeaders = ClassUtils.getMethod(builder, "copyHeaders", Map.class);
Method build = ClassUtils.getMethod(builder, "build");
Object instance = ReflectionUtils.invokeMethod(withPayload, null, payload);
ReflectionUtils.invokeMethod(copyHeaders, instance, headers);
return ReflectionUtils.invokeMethod(build, instance);
}
/**
* Convert a message from the handler into one that is safe to consume in the caller's
* class loader. If the handler is a wrapper for a function in an isolated class
* loader, then the message will be created with the target class loader (therefore
* the {@link Message} class must be on the classpath of the target class loader).
* @param handler the function that generated the message
* @param message the message to convert
* @return a message with the correct class loader
*/
public static Message<?> unpack(Object handler, Object message) {
if (handler instanceof FluxWrapper) {
handler = ((FluxWrapper<?>) handler).getTarget();
}
if (!(handler instanceof Isolated)) {
if (message instanceof Message) {
return (Message<?>) message;
}
return MessageBuilder.withPayload(message).build();
}
ClassLoader classLoader = ((Isolated) handler).getClassLoader();
Class<?> type = ClassUtils.isPresent(Message.class.getName(), classLoader)
? ClassUtils.resolveClassName(Message.class.getName(), classLoader)
: null;
Object payload;
Map<String, Object> headers;
if (type != null && type.isAssignableFrom(message.getClass())) {
Method getPayload = ClassUtils.getMethod(type, "getPayload");
Method getHeaders = ClassUtils.getMethod(type, "getHeaders");
payload = ReflectionUtils.invokeMethod(getPayload, message);
@SuppressWarnings("unchecked")
Map<String, Object> map = (Map<String, Object>) ReflectionUtils
.invokeMethod(getHeaders, message);
headers = map;
}
else {
payload = message;
headers = Collections.emptyMap();
}
return MessageBuilder.withPayload(payload).copyHeaders(headers).build();
}
}

58
spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/FunctionRegistrationTests.java
View File

@@ -1,58 +0,0 @@
/*
* Copyright 2012-2019 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
*
* https://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.function.context;
import java.util.function.Function;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Dave Syer
*
*/
public class FunctionRegistrationTests {
@Test
public void noTypeByDefault() {
FunctionRegistration<?> registration = new FunctionRegistration<>(new Foos(),
"foos");
assertThat(registration.getType()).isNull();
assertThat(registration.getNames()).contains("foos");
}
@Test
public void wrap() {
FunctionRegistration<Foos> registration = new FunctionRegistration<>(new Foos(),
"foos").type(FunctionType.of(Foos.class).getType());
FunctionRegistration<?> other = registration.wrap();
assertThat(registration.getType().isWrapper()).isFalse();
assertThat(other.getType().isWrapper()).isTrue();
assertThat(other.getTarget()).isNotEqualTo(registration.getTarget());
}
private static class Foos implements Function<Integer, String> {
@Override
public String apply(Integer t) {
return "i=" + t;
}
}
}

346
spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/FunctionTypeTests.java
View File

@@ -1,346 +0,0 @@
/*
* Copyright 2012-2019 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
*
* https://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.function.context;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.Collections;
import java.util.Map;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuple3;
import org.springframework.core.ResolvableType;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Dave Syer
*
*/
public class FunctionTypeTests {
@Test
public void functionWithTuples() {
FunctionType functionType = FunctionType.of(MyFunction.class);
assertThat(functionType.getType()).isInstanceOf(ParameterizedType.class);
}
@Test
public void plainFunction() {
FunctionType function = new FunctionType(IntegerToString.class);
assertThat(function.getInputType()).isEqualTo(Integer.class);
assertThat(function.getOutputType()).isEqualTo(String.class);
assertThat(function.getInputWrapper()).isEqualTo(Integer.class);
assertThat(function.getOutputWrapper()).isEqualTo(String.class);
assertThat(function.isMessage()).isEqualTo(false);
}
@Test
public void supplierOfRegistration() {
FunctionType function = new FunctionType(
SupplierOfRegistrationOfIntegerToString.class);
assertThat(function.getInputType()).isEqualTo(Integer.class);
assertThat(function.getOutputType()).isEqualTo(String.class);
assertThat(function.getInputWrapper()).isEqualTo(Integer.class);
assertThat(function.getOutputWrapper()).isEqualTo(String.class);
assertThat(function.isMessage()).isEqualTo(false);
}
@Test
public void supplier() {
FunctionType function = new FunctionType(SupplierOfIntegerToString.class);
assertThat(function.getInputType()).isEqualTo(Integer.class);
assertThat(function.getOutputType()).isEqualTo(String.class);
assertThat(function.getInputWrapper()).isEqualTo(Integer.class);
assertThat(function.getOutputWrapper()).isEqualTo(String.class);
assertThat(function.isMessage()).isEqualTo(false);
}
@Test
public void genericFunction() {
FunctionType function = new FunctionType(StringToMap.class);
assertThat(function.getInputType()).isEqualTo(String.class);
assertThat(function.getOutputType()).isEqualTo(Map.class);
assertThat(function.getInputWrapper()).isEqualTo(String.class);
assertThat(function.getOutputWrapper()).isEqualTo(Map.class);
assertThat(function.isMessage()).isEqualTo(false);
}
@Test
public void pojoFunction() {
FunctionType function = new FunctionType(FooToFoo.class);
assertThat(function.getInputType()).isEqualTo(Foo.class);
assertThat(function.getOutputType()).isEqualTo(Bar.class);
assertThat(function.getInputWrapper()).isEqualTo(Foo.class);
assertThat(function.getOutputWrapper()).isEqualTo(Bar.class);
assertThat(function.isMessage()).isEqualTo(false);
}
@Test
public void fluxFunction() {
FunctionType function = new FunctionType(FluxToFlux.class);
assertThat(function.getInputType()).isEqualTo(Foo.class);
assertThat(function.getOutputType()).isEqualTo(Bar.class);
assertThat(function.getInputWrapper()).isEqualTo(Flux.class);
assertThat(function.getOutputWrapper()).isEqualTo(Flux.class);
assertThat(function.isMessage()).isEqualTo(false);
}
@Test
public void fluxMessageFunction() {
FunctionType function = new FunctionType(FluxMessageToFluxMessage.class);
assertThat(function.getInputType()).isEqualTo(Foo.class);
assertThat(function.getOutputType()).isEqualTo(Bar.class);
assertThat(function.getInputWrapper()).isEqualTo(Flux.class);
assertThat(function.getOutputWrapper()).isEqualTo(Flux.class);
assertThat(function.isMessage()).isEqualTo(true);
}
@Test
public void plainFunctionFromType() {
Type type = ResolvableType
.forClassWithGenerics(Function.class, Integer.class, String.class)
.getType();
FunctionType function = new FunctionType(type);
assertThat(function.getInputType()).isEqualTo(Integer.class);
assertThat(function.getOutputType()).isEqualTo(String.class);
assertThat(function.getInputWrapper()).isEqualTo(Integer.class);
assertThat(function.getOutputWrapper()).isEqualTo(String.class);
assertThat(function.isMessage()).isEqualTo(false);
}
@Test
public void pojoConsumerFromType() {
Type type = ResolvableType.forClassWithGenerics(Consumer.class, Foo.class)
.getType();
FunctionType function = new FunctionType(type);
assertThat(function.getInputType()).isEqualTo(Foo.class);
assertThat(function.getOutputType()).isEqualTo(Void.class);
assertThat(function.getInputWrapper()).isEqualTo(Foo.class);
assertThat(function.getOutputWrapper()).isEqualTo(Void.class);
assertThat(function.isMessage()).isEqualTo(false);
}
@Test
public void pojoSupplierFromType() {
Type type = ResolvableType.forClassWithGenerics(Supplier.class, Foo.class)
.getType();
FunctionType function = new FunctionType(type);
assertThat(function.getInputType()).isEqualTo(Void.class);
assertThat(function.getOutputType()).isEqualTo(Foo.class);
assertThat(function.getInputWrapper()).isEqualTo(Void.class);
assertThat(function.getOutputWrapper()).isEqualTo(Foo.class);
assertThat(function.isMessage()).isEqualTo(false);
}
@Test
public void pojoSupplierFrom() {
FunctionType function = new FunctionType(Supplier.class).to(Foo.class);
assertThat(function.getInputType()).isEqualTo(Void.class);
assertThat(function.getOutputType()).isEqualTo(Foo.class);
assertThat(function.getInputWrapper()).isEqualTo(Void.class);
assertThat(function.getOutputWrapper()).isEqualTo(Foo.class);
assertThat(function.isMessage()).isEqualTo(false);
}
@Test
public void pojoSupplier() {
FunctionType function = FunctionType.supplier(Foo.class);
assertThat(function.getInputType()).isEqualTo(Void.class);
assertThat(function.getOutputType()).isEqualTo(Foo.class);
assertThat(function.getInputWrapper()).isEqualTo(Void.class);
assertThat(function.getOutputWrapper()).isEqualTo(Foo.class);
assertThat(function.isMessage()).isEqualTo(false);
}
@Test
public void pojoConsumer() {
FunctionType function = FunctionType.consumer(Foo.class);
assertThat(function.getOutputType()).isEqualTo(Void.class);
assertThat(function.getInputType()).isEqualTo(Foo.class);
assertThat(function.getOutputWrapper()).isEqualTo(Void.class);
assertThat(function.getInputWrapper()).isEqualTo(Foo.class);
assertThat(function.isMessage()).isEqualTo(false);
}
@Test
public void plainFunctionFromApi() {
FunctionType function = FunctionType.from(Integer.class).to(String.class);
assertThat(function.getInputType()).isEqualTo(Integer.class);
assertThat(function.getOutputType()).isEqualTo(String.class);
assertThat(function.getInputWrapper()).isEqualTo(Integer.class);
assertThat(function.getOutputWrapper()).isEqualTo(String.class);
assertThat(function.isMessage()).isEqualTo(false);
}
@Test
public void fluxMessageFunctionFromType() {
Type type = ResolvableType
.forClassWithGenerics(Function.class,
ResolvableType.forClassWithGenerics(
Flux.class,
ResolvableType.forClassWithGenerics(Message.class,
Foo.class)),
ResolvableType.forClassWithGenerics(Flux.class, ResolvableType
.forClassWithGenerics(Message.class, Bar.class)))
.getType();
FunctionType function = new FunctionType(type);
assertThat(function.getInputType()).isEqualTo(Foo.class);
assertThat(function.getOutputType()).isEqualTo(Bar.class);
assertThat(function.getInputWrapper()).isEqualTo(Flux.class);
assertThat(function.getOutputWrapper()).isEqualTo(Flux.class);
assertThat(function.isMessage()).isEqualTo(true);
}
@Test
public void fluxMessageFunctionFromApi() {
FunctionType function = FunctionType.from(Foo.class).to(Bar.class).message()
.wrap(Flux.class);
assertThat(function.getInputType()).isEqualTo(Foo.class);
assertThat(function.getOutputType()).isEqualTo(Bar.class);
assertThat(function.getInputWrapper()).isEqualTo(Flux.class);
assertThat(function.getOutputWrapper()).isEqualTo(Flux.class);
assertThat(function.isMessage()).isEqualTo(true);
}
@Test
public void compose() {
FunctionType input = FunctionType.from(Foo.class).to(Bar.class).wrap(Flux.class);
FunctionType output = FunctionType.from(Bar.class).to(String.class);
FunctionType function = FunctionType.compose(input, output);
assertThat(function.getInputType()).isEqualTo(Foo.class);
assertThat(function.getOutputType()).isEqualTo(String.class);
assertThat(function.getInputWrapper()).isEqualTo(Flux.class);
assertThat(function.getOutputWrapper()).isEqualTo(Flux.class);
assertThat(function.isMessage()).isEqualTo(false);
}
@Test
public void idempotentMessage() {
FunctionType function = FunctionType.from(Foo.class).to(Bar.class).message()
.wrap(Flux.class);
assertThat(function).isSameAs(function.message());
}
@Test
public void idempotentWrapper() {
FunctionType function = FunctionType.from(Foo.class).to(Bar.class).message()
.wrap(Flux.class);
assertThat(function).isSameAs(function.wrap(Flux.class));
}
@Test
public void nonWrapper() {
FunctionType function = FunctionType.from(Foo.class).to(Bar.class);
assertThat(function).isSameAs(function.wrap(Object.class));
}
private static class SupplierOfRegistrationOfIntegerToString
implements Supplier<FunctionRegistration<Function<Integer, String>>> {
@Override
public FunctionRegistration<Function<Integer, String>> get() {
return new FunctionRegistration<Function<Integer, String>>(
new IntegerToString(), "ints");
}
}
private static class SupplierOfIntegerToString
implements Supplier<Function<Integer, String>> {
@Override
public Function<Integer, String> get() {
return new IntegerToString();
}
}
private static class IntegerToString implements Function<Integer, String> {
@Override
public String apply(Integer t) {
return "" + t;
}
}
private static class StringToMap implements Function<String, Map<String, Integer>> {
@Override
public Map<String, Integer> apply(String t) {
return Collections.emptyMap();
}
}
private static class FooToFoo implements Function<Foo, Bar> {
@Override
public Bar apply(Foo t) {
return new Bar();
}
}
private static class FluxToFlux implements Function<Flux<Foo>, Flux<Bar>> {
@Override
public Flux<Bar> apply(Flux<Foo> t) {
return t.map(f -> new Bar());
}
}
private static class FluxMessageToFluxMessage
implements Function<Flux<Message<Foo>>, Flux<Message<Bar>>> {
@Override
public Flux<Message<Bar>> apply(Flux<Message<Foo>> t) {
return t.map(f -> MessageBuilder.withPayload(new Bar())
.copyHeadersIfAbsent(f.getHeaders()).build());
}
}
private static class Foo {
}
private static class Bar {
}
private static class MyFunction
implements Function<Tuple2<Flux<String>, Flux<Integer>>, Tuple3<Flux<String>, Flux<Integer>, Flux<Object>>> {
@Override
public Tuple3<Flux<String>, Flux<Integer>, Flux<Object>> apply(Tuple2<Flux<String>, Flux<Integer>> t) {
return null;
}
}
}

239
spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/SpringFunctionAdapterInitializerTests.java
View File

@@ -1,239 +0,0 @@
/*
* Copyright 2019-2019 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
*
* https://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.function.context;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import org.springframework.cloud.function.context.config.ContextFunctionCatalogAutoConfiguration;
import org.springframework.context.ApplicationContextInitializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.context.support.GenericApplicationContext;
import static org.assertj.core.api.Assertions.assertThat;
/**
*
* @author Oleg Zhurakousky
*
*/
public class SpringFunctionAdapterInitializerTests {
private AbstractSpringFunctionAdapterInitializer<Object> initializer;
@AfterEach
public void after() {
System.clearProperty("function.name");
if (this.initializer != null) {
this.initializer.close();
}
}
@Test
public void nullSource() {
Assertions.assertThrows(IllegalArgumentException.class, () ->
this.initializer = new AbstractSpringFunctionAdapterInitializer<Object>(null) {
});
}
@Test
public void sourceAsMainClassProperty() {
try {
System.setProperty("MAIN_CLASS", FluxFunctionConfig.class.getName());
this.initializer = new AbstractSpringFunctionAdapterInitializer<Object>() {
};
}
finally {
System.clearProperty("MAIN_CLASS");
}
}
@Test
public void functionBean() {
this.initializer = new AbstractSpringFunctionAdapterInitializer<Object>(FluxFunctionConfig.class) {
};
this.initializer.initialize(null);
Flux<?> result = Flux.from(this.initializer.apply(Flux.just(new Foo())));
assertThat(result.blockFirst()).isInstanceOf(Bar.class);
}
@Test
public void functionApp() {
this.initializer = new AbstractSpringFunctionAdapterInitializer<Object>(FluxFunctionApp.class) {
};
this.initializer.initialize(null);
Flux<?> result = Flux.from(this.initializer.apply(Flux.just(new Foo())));
Object o = result.blockFirst();
assertThat(result.blockFirst()).isInstanceOf(Bar.class);
}
@Test
public void functionCatalog() {
this.initializer = new AbstractSpringFunctionAdapterInitializer<Object>(FunctionConfig.class) {
};
this.initializer.initialize(null);
Flux<?> result = Flux.from(this.initializer.apply(Flux.just(new Foo())));
assertThat(result.blockFirst()).isInstanceOf(Bar.class);
}
@Test
@Disabled // related to boot 2.1 no bean override change
public void functionRegistrar() {
this.initializer = new AbstractSpringFunctionAdapterInitializer<Object>(FunctionRegistrar.class) {
};
this.initializer.initialize(null);
Flux<?> result = Flux.from(this.initializer.apply(Flux.just(new Foo())));
assertThat(result.blockFirst()).isInstanceOf(Bar.class);
}
@Test
public void namedFunctionCatalog() {
this.initializer = new AbstractSpringFunctionAdapterInitializer<Object>(NamedFunctionConfig.class) {
};
System.setProperty("function.name", "other");
this.initializer.initialize(null);
Flux<?> result = Flux.from(this.initializer.apply(Flux.just(new Foo())));
assertThat(result.blockFirst()).isInstanceOf(Bar.class);
}
@Test
public void consumerCatalog() {
this.initializer = new AbstractSpringFunctionAdapterInitializer<Object>(ConsumerConfig.class) {
};
this.initializer.initialize(null);
Flux<?> result = Flux.from(this.initializer.apply(Flux.just(new Foo())));
assertThat(result.toStream().collect(Collectors.toList())).isEmpty();
}
@Test
public void supplierCatalog() {
initializer = new AbstractSpringFunctionAdapterInitializer<Object>(SupplierConfig.class) {
};
initializer.initialize(null);
Flux result = Flux.from(initializer.apply(null));
assertThat(result.blockFirst()).isInstanceOf(Bar.class);
}
@Configuration
protected static class FluxFunctionConfig {
@Bean
public Function<Flux<Foo>, Flux<Bar>> function() {
return flux -> flux.map(foo -> new Bar());
}
}
protected static class FluxFunctionApp implements Function<Flux<Foo>, Flux<Bar>> {
@Override
public Flux<Bar> apply(Flux<Foo> flux) {
return flux.map(foo -> new Bar());
}
}
protected static class FunctionRegistrar
implements ApplicationContextInitializer<GenericApplicationContext> {
public Function<Flux<Foo>, Flux<Bar>> function() {
return flux -> flux.map(foo -> new Bar());
}
@Override
public void initialize(GenericApplicationContext context) {
context.registerBean("function", FunctionRegistration.class,
() -> new FunctionRegistration<Function<Flux<Foo>, Flux<Bar>>>(
function()).name("function")
.type(FunctionType.from(Foo.class).to(Bar.class)
.wrap(Flux.class).getType()));
}
}
@Configuration
@Import(ContextFunctionCatalogAutoConfiguration.class)
protected static class FunctionConfig {
@Bean
public Function<Foo, Bar> function() {
return foo -> new Bar();
}
}
@Configuration
@Import(ContextFunctionCatalogAutoConfiguration.class)
protected static class NamedFunctionConfig {
@Bean
public Function<Foo, Bar> other() {
return foo -> new Bar();
}
}
@Configuration
@Import(ContextFunctionCatalogAutoConfiguration.class)
protected static class SupplierConfig {
@Bean
public Supplier<Bar> supplier() {
return () -> {
return new Bar();
};
}
}
@Configuration
@Import(ContextFunctionCatalogAutoConfiguration.class)
protected static class ConsumerConfig {
@Bean
public Consumer<Foo> consumer() {
return foo -> {
};
}
}
protected static class Foo {
}
protected static class Bar {
}
}

20
spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistryTests.java
View File

@@ -53,7 +53,6 @@ import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.function.context.FunctionCatalog;
import org.springframework.cloud.function.context.FunctionRegistration;
import org.springframework.cloud.function.context.FunctionRegistry;
import org.springframework.cloud.function.context.FunctionType;
import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
import org.springframework.cloud.function.json.JsonMapper;
import org.springframework.context.ApplicationContext;
@@ -499,30 +498,35 @@ public class BeanFactoryAwareFunctionRegistryTests {
assertThat(func).isNull();
FunctionRegistry registry = (FunctionRegistry) catalog;
try {
FunctionRegistration registration = new FunctionRegistration(new MyFunction(), "a").type(FunctionType.from(Integer.class).to(String.class));
FunctionRegistration registration = new FunctionRegistration(new MyFunction(), "a")
.type(FunctionTypeUtils.functionType(Integer.class, String.class));
registry.register(registration);
fail();
}
catch (IllegalStateException e) {
catch (IllegalArgumentException e) {
// good as we expect it to fail
}
//
try {
FunctionRegistration registration = new FunctionRegistration(new MyFunction(), "b").type(FunctionType.from(String.class).to(Integer.class));
FunctionRegistration registration = new FunctionRegistration(new MyFunction(), "b")
.type(FunctionTypeUtils.functionType(String.class, Integer.class));
registry.register(registration);
fail();
}
catch (IllegalStateException e) {
catch (IllegalArgumentException e) {
// good as we expect it to fail
}
//
FunctionRegistration c = new FunctionRegistration(new MyFunction(), "c").type(FunctionType.from(String.class).to(String.class));
FunctionRegistration c = new FunctionRegistration(new MyFunction(), "c")
.type(FunctionTypeUtils.functionType(String.class, String.class));
registry.register(c);
//
FunctionRegistration d = new FunctionRegistration(new RawFunction(), "d").type(FunctionType.from(Person.class).to(String.class));
FunctionRegistration d = new FunctionRegistration(new RawFunction(), "d")
.type(FunctionTypeUtils.functionType(Person.class, String.class));
registry.register(d);
//
FunctionRegistration e = new FunctionRegistration(new RawFunction(), "e").type(FunctionType.from(Object.class).to(Object.class));
FunctionRegistration e = new FunctionRegistration(new RawFunction(), "e")
.type(FunctionTypeUtils.functionType(Object.class, Object.class));
registry.register(e);
}

2
spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwarePojoFunctionRegistryTests.java
View File

@@ -133,7 +133,7 @@ public class BeanFactoryAwarePojoFunctionRegistryTests {
}
// POJO Function
private static class MyFunctionLike {
public static class MyFunctionLike {
public String uppercase(String value) {
return value.toUpperCase();
}

31
spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/FunctionTypeUtilsTests.java
View File

@@ -31,7 +31,6 @@ import reactor.core.publisher.Mono;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuple3;
import org.springframework.cloud.function.context.FunctionType;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.messaging.Message;
@@ -114,29 +113,29 @@ public class FunctionTypeUtilsTests {
@Test
public void testFunctionTypeByClassDiscovery() {
FunctionType type = FunctionType.of(FunctionTypeUtils.discoverFunctionTypeFromClass(Function.class));
assertThat(type.getInputType()).isAssignableFrom(Object.class);
Type type = FunctionTypeUtils.discoverFunctionTypeFromClass(Function.class);
assertThat(FunctionTypeUtils.getRawType(FunctionTypeUtils.getInputType(type))).isAssignableFrom(Object.class);
type = FunctionType.of(FunctionTypeUtils.discoverFunctionTypeFromClass(MessageFunction.class));
assertThat(type.getInputType()).isAssignableFrom(String.class);
assertThat(type.getOutputType()).isAssignableFrom(String.class);
type = FunctionTypeUtils.discoverFunctionTypeFromClass(MessageFunction.class);
assertThat(FunctionTypeUtils.getRawType(FunctionTypeUtils.getComponentTypeOfInputType(type))).isAssignableFrom(String.class);
assertThat(FunctionTypeUtils.getRawType(FunctionTypeUtils.getComponentTypeOfOutputType(type))).isAssignableFrom(String.class);
type = FunctionType.of(FunctionTypeUtils.discoverFunctionTypeFromClass(MyMessageFunction.class));
assertThat(type.getInputType()).isAssignableFrom(String.class);
assertThat(type.getOutputType()).isAssignableFrom(String.class);
type = FunctionTypeUtils.discoverFunctionTypeFromClass(MyMessageFunction.class);
assertThat(FunctionTypeUtils.getRawType(FunctionTypeUtils.getComponentTypeOfInputType(type))).isAssignableFrom(String.class);
assertThat(FunctionTypeUtils.getRawType(FunctionTypeUtils.getComponentTypeOfOutputType(type))).isAssignableFrom(String.class);
type = FunctionType.of(FunctionTypeUtils.discoverFunctionTypeFromClass(MessageConsumer.class));
assertThat(type.getInputType()).isAssignableFrom(String.class);
type = FunctionTypeUtils.discoverFunctionTypeFromClass(MessageConsumer.class);
assertThat(FunctionTypeUtils.getRawType(FunctionTypeUtils.getComponentTypeOfInputType(type))).isAssignableFrom(String.class);
type = FunctionType.of(FunctionTypeUtils.discoverFunctionTypeFromClass(MyMessageConsumer.class));
assertThat(type.getInputType()).isAssignableFrom(String.class);
type = FunctionTypeUtils.discoverFunctionTypeFromClass(MyMessageConsumer.class);
assertThat(FunctionTypeUtils.getRawType(FunctionTypeUtils.getComponentTypeOfInputType(type))).isAssignableFrom(String.class);
}
@Test
public void testWithComplexHierarchy() {
FunctionType type = FunctionType.of(FunctionTypeUtils.discoverFunctionTypeFromClass(ReactiveFunctionImpl.class));
assertThat(String.class).isAssignableFrom(type.getInputType());
assertThat(Integer.class).isAssignableFrom(type.getOutputType());
Type type = FunctionTypeUtils.discoverFunctionTypeFromClass(ReactiveFunctionImpl.class);
assertThat(String.class).isAssignableFrom(FunctionTypeUtils.getRawType(FunctionTypeUtils.getComponentTypeOfInputType(type)));
assertThat(Integer.class).isAssignableFrom(FunctionTypeUtils.getRawType(FunctionTypeUtils.getComponentTypeOfOutputType(type)));
}
@Test

49
spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/MessageConsumerTests.java
View File

@@ -1,49 +0,0 @@
/*
* Copyright 2019-2019 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
*
* https://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.function.context.catalog;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Consumer;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Dave Syer
*/
public class MessageConsumerTests {
private List<String> items = new ArrayList<>();
@Test
public void plainConsumer() {
MessageConsumer consumer = new MessageConsumer(input());
consumer.accept(Flux
.just(MessageBuilder.withPayload("foo").setHeader("foo", "bar").build()));
assertThat(this.items).hasSize(1);
}
private Consumer<String> input() {
return value -> this.items.add(value);
}
}

113
spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/MessageFunctionTests.java
View File

@@ -1,113 +0,0 @@
/*
* Copyright 2019-2019 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
*
* https://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.function.context.catalog;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Consumer;
import java.util.function.Function;
import org.junit.jupiter.api.Test;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import org.springframework.cloud.function.core.FluxConsumer;
import org.springframework.cloud.function.core.FluxFunction;
import org.springframework.cloud.function.core.FluxToMonoFunction;
import org.springframework.cloud.function.core.MonoToFluxFunction;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Dave Syer
*/
public class MessageFunctionTests {
private List<String> items = new ArrayList<>();
@Test
public void plainFunction() {
MessageFunction function = new MessageFunction(uppercase());
Publisher<Message<?>> result = function.apply(Flux
.just(MessageBuilder.withPayload("foo").setHeader("foo", "bar").build()));
StepVerifier.create(result).assertNext(message -> {
assertThat(message.getPayload()).isEqualTo("FOO");
assertThat(message.getHeaders()).containsEntry("foo", "bar");
});
}
@Test
public void fluxFunction() {
MessageFunction function = new MessageFunction(new FluxFunction<>(uppercase()));
Publisher<Message<?>> result = function.apply(Flux
.just(MessageBuilder.withPayload("foo").setHeader("foo", "bar").build()));
StepVerifier.create(result).assertNext(message -> {
assertThat(message.getPayload()).isEqualTo("FOO");
assertThat(message.getHeaders()).containsEntry("foo", "bar");
});
}
@Test
public void fluxToMonoFunction() {
MessageFunction function = new MessageFunction(
new FluxToMonoFunction<String, String>(
flux -> flux.next().map(uppercase())));
Publisher<Message<?>> result = function.apply(Flux
.just(MessageBuilder.withPayload("foo").setHeader("foo", "bar").build()));
StepVerifier.create(result).assertNext(message -> {
assertThat(message.getPayload()).isEqualTo("FOO");
assertThat(message.getHeaders()).containsEntry("foo", "bar");
});
}
@Test
public void monoToFunction() {
MessageFunction function = new MessageFunction(
new MonoToFluxFunction<String, String>(
mono -> Flux.from(mono.map(uppercase()))));
Publisher<Message<?>> result = function.apply(Flux
.just(MessageBuilder.withPayload("foo").setHeader("foo", "bar").build()));
StepVerifier.create(result).assertNext(message -> {
assertThat(message.getPayload()).isEqualTo("FOO");
assertThat(message.getHeaders()).containsEntry("foo", "bar");
});
}
@Test
public void fluxConsumer() {
MessageFunction function = new MessageFunction(new FluxConsumer<>(stash()));
Publisher<Message<?>> result = function.apply(Flux
.just(MessageBuilder.withPayload("foo").setHeader("foo", "bar").build()));
StepVerifier.create(result).assertNext(message -> {
assertThat(message.getPayload()).isEqualTo(null);
assertThat(message.getHeaders()).containsEntry("foo", "bar");
assertThat(this.items).hasSize(1);
});
}
private Consumer<String> stash() {
return value -> this.items.add(value);
}
private Function<String, String> uppercase() {
return value -> value.toUpperCase();
}
}

79
spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/MessageSupplierTests.java
View File

@@ -1,79 +0,0 @@
/*
* Copyright 2019-2019 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
*
* https://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.function.context.catalog;
import java.util.Arrays;
import java.util.Collection;
import java.util.function.Supplier;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Dave Syer
*/
public class MessageSupplierTests {
@Test
public void plainSupplier() {
MessageSupplier supplier = new MessageSupplier(input());
StepVerifier.create(supplier.get()).assertNext(message -> {
assertThat(message.getPayload()).isEqualTo("foo");
assertThat(message.getHeaders()).isEmpty();
});
}
@Test
public void collectionSupplier() {
MessageSupplier supplier = new MessageSupplier(inputs());
StepVerifier.create(supplier.get()).assertNext(message -> {
assertThat(message.getPayload()).isEqualTo("foo");
assertThat(message.getHeaders()).isEmpty();
}).assertNext(message -> {
assertThat(message.getPayload()).isEqualTo("bar");
assertThat(message.getHeaders()).isEmpty();
});
}
@Test
public void fluxSupplier() {
MessageSupplier supplier = new MessageSupplier(flux());
StepVerifier.create(supplier.get()).assertNext(message -> {
assertThat(message.getPayload()).isEqualTo("foo");
assertThat(message.getHeaders()).isEmpty();
}).assertNext(message -> {
assertThat(message.getPayload()).isEqualTo("bar");
assertThat(message.getHeaders()).isEmpty();
});
}
private Supplier<String> input() {
return () -> "foo";
}
private Supplier<Collection<String>> inputs() {
return () -> Arrays.asList("foo", "bar");
}
private Supplier<Flux<String>> flux() {
return () -> Flux.just("foo", "bar");
}
}

77
spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistryTests.java
View File

@@ -43,7 +43,6 @@ import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.function.context.FunctionCatalog;
import org.springframework.cloud.function.context.FunctionRegistration;
import org.springframework.cloud.function.context.FunctionRegistry;
import org.springframework.cloud.function.context.FunctionType;
import org.springframework.cloud.function.context.HybridFunctionalRegistrationTests.UppercaseFunction;
import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
import org.springframework.cloud.function.context.config.JsonMessageConverter;
@@ -94,7 +93,7 @@ public class SimpleFunctionRegistryTests {
public void testCachingOfFunction() {
Echo function = new Echo();
FunctionRegistration<Echo> registration = new FunctionRegistration<>(
function, "echo").type(FunctionType.of(Echo.class));
function, "echo").type(Echo.class);
SimpleFunctionRegistry catalog = new SimpleFunctionRegistry(this.conversionService, this.messageConverter,
new JacksonMapper(new ObjectMapper()));
catalog.register(registration);
@@ -110,7 +109,7 @@ public class SimpleFunctionRegistryTests {
public void testNoCachingOfFunction() {
Echo function = new Echo();
FunctionRegistration<Echo> registration = new FunctionRegistration<>(
function, "echo").type(FunctionType.of(Echo.class));
function, "echo").type(Echo.class);
registration.getProperties().put("singleton", "false");
SimpleFunctionRegistry catalog = new SimpleFunctionRegistry(this.conversionService, this.messageConverter,
new JacksonMapper(new ObjectMapper()));
@@ -136,7 +135,7 @@ public class SimpleFunctionRegistryTests {
};
FunctionRegistration<Function<Map<String, Person>, String>> registration = new FunctionRegistration<>(
function, "echo").type(FunctionType.of(functionType));
function, "echo").type(functionType);
SimpleFunctionRegistry catalog = new SimpleFunctionRegistry(this.conversionService, this.messageConverter,
new JacksonMapper(new ObjectMapper()));
catalog.register(registration);
@@ -150,7 +149,7 @@ public class SimpleFunctionRegistryTests {
public void testSCF640() {
Echo function = new Echo();
FunctionRegistration<Echo> registration = new FunctionRegistration<>(
function, "echo").type(FunctionType.of(Echo.class));
function, "echo").type(Echo.class);
SimpleFunctionRegistry catalog = new SimpleFunctionRegistry(this.conversionService, this.messageConverter,
new JacksonMapper(new ObjectMapper()));
catalog.register(registration);
@@ -168,27 +167,27 @@ public class SimpleFunctionRegistryTests {
new JacksonMapper(new ObjectMapper()));
FunctionRegistration<UpperCase> reg1 = new FunctionRegistration<>(
new UpperCase(), "uppercase").type(FunctionType.of(UpperCase.class));
new UpperCase(), "uppercase").type(UpperCase.class);
catalog.register(reg1);
//
FunctionRegistration<UpperCaseMessage> reg2 = new FunctionRegistration<>(
new UpperCaseMessage(), "uppercaseMessage").type(FunctionType.of(UpperCaseMessage.class));
new UpperCaseMessage(), "uppercaseMessage").type(UpperCaseMessage.class);
catalog.register(reg2);
//
FunctionRegistration<StringArrayFunction> reg3 = new FunctionRegistration<>(
new StringArrayFunction(), "stringArray").type(FunctionType.of(StringArrayFunction.class));
new StringArrayFunction(), "stringArray").type(StringArrayFunction.class);
catalog.register(reg3);
//
FunctionRegistration<TypelessFunction> reg4 = new FunctionRegistration<>(
new TypelessFunction(), "typeless").type(FunctionType.of(TypelessFunction.class));
new TypelessFunction(), "typeless").type(TypelessFunction.class);
catalog.register(reg4);
//
FunctionRegistration<ByteArrayFunction> reg5 = new FunctionRegistration<>(
new ByteArrayFunction(), "typeless").type(FunctionType.of(ByteArrayFunction.class));
new ByteArrayFunction(), "typeless").type(ByteArrayFunction.class);
catalog.register(reg5);
//
FunctionRegistration<StringListFunction> reg6 = new FunctionRegistration<>(
new StringListFunction(), "stringList").type(FunctionType.of(StringListFunction.class));
new StringListFunction(), "stringList").type(StringListFunction.class);
catalog.register(reg6);
Message<String> collectionMessage = MessageBuilder.withPayload("[\"ricky\", \"julien\", \"bubbles\"]").build();
@@ -244,7 +243,7 @@ public class SimpleFunctionRegistryTests {
UpperCase function = new UpperCase();
FunctionRegistration<UpperCase> registration = new FunctionRegistration<>(
function, "foo").type(FunctionType.of(UppercaseFunction.class));
function, "foo").type(UppercaseFunction.class);
SimpleFunctionRegistry catalog = new SimpleFunctionRegistry(this.conversionService, this.messageConverter,
new JacksonMapper(new ObjectMapper()));
catalog.register(registration);
@@ -263,7 +262,7 @@ public class SimpleFunctionRegistryTests {
TestFunction function = new TestFunction();
FunctionRegistration<TestFunction> registration = new FunctionRegistration<>(
function, "foo").type(FunctionType.of(TestFunction.class));
function, "foo").type(TestFunction.class);
SimpleFunctionRegistry catalog = new SimpleFunctionRegistry(this.conversionService, this.messageConverter,
new JacksonMapper(new ObjectMapper()));
catalog.register(registration);
@@ -272,7 +271,7 @@ public class SimpleFunctionRegistryTests {
FunctionInvocationWrapper lookedUpFunction = catalog.lookup("hello");
assertThat(lookedUpFunction).isNotNull(); // because we only have one and can look it up with any name
FunctionRegistration<TestFunction> registration2 = new FunctionRegistration<>(
function, "foo2").type(FunctionType.of(TestFunction.class));
function, "foo2").type(TestFunction.class);
catalog.register(registration2);
lookedUpFunction = catalog.lookup("hello");
assertThat(lookedUpFunction).isNull();
@@ -283,9 +282,9 @@ public class SimpleFunctionRegistryTests {
@Test
public void testFunctionComposition() {
FunctionRegistration<UpperCase> upperCaseRegistration = new FunctionRegistration<>(
new UpperCase(), "uppercase").type(FunctionType.of(UpperCase.class));
new UpperCase(), "uppercase").type(UpperCase.class);
FunctionRegistration<Reverse> reverseRegistration = new FunctionRegistration<>(
new Reverse(), "reverse").type(FunctionType.of(Reverse.class));
new Reverse(), "reverse").type(Reverse.class);
SimpleFunctionRegistry catalog = new SimpleFunctionRegistry(this.conversionService, this.messageConverter,
new JacksonMapper(new ObjectMapper()));
catalog.register(upperCaseRegistration);
@@ -294,23 +293,15 @@ public class SimpleFunctionRegistryTests {
Function<Flux<String>, Flux<String>> lookedUpFunction = catalog
.lookup("uppercase|reverse");
assertThat(lookedUpFunction).isNotNull();
Flux flux = lookedUpFunction.apply(Flux.just("star"));
flux.subscribe(v -> {
System.out.println(v);
});
// assertThat(lookedUpFunction.apply(Flux.just("star")).blockFirst())
// .isEqualTo("RATS");
}
@Test
@Disabled
public void testFunctionCompositionImplicit() {
FunctionRegistration<Words> wordsRegistration = new FunctionRegistration<>(
new Words(), "words").type(FunctionType.of(Words.class));
new Words(), "words").type(Words.class);
FunctionRegistration<Reverse> reverseRegistration = new FunctionRegistration<>(
new Reverse(), "reverse").type(FunctionType.of(Reverse.class));
new Reverse(), "reverse").type(Reverse.class);
FunctionRegistry catalog = new SimpleFunctionRegistry(this.conversionService, this.messageConverter,
new JacksonMapper(new ObjectMapper()));
catalog.register(wordsRegistration);
@@ -327,9 +318,9 @@ public class SimpleFunctionRegistryTests {
@Disabled
public void testFunctionCompletelyImplicitComposition() {
FunctionRegistration<Words> wordsRegistration = new FunctionRegistration<>(
new Words(), "words").type(FunctionType.of(Words.class));
new Words(), "words").type(Words.class);
FunctionRegistration<Reverse> reverseRegistration = new FunctionRegistration<>(
new Reverse(), "reverse").type(FunctionType.of(Reverse.class));
new Reverse(), "reverse").type(Reverse.class);
SimpleFunctionRegistry catalog = new SimpleFunctionRegistry(this.conversionService, this.messageConverter,
new JacksonMapper(new ObjectMapper()));
catalog.register(wordsRegistration);
@@ -345,9 +336,9 @@ public class SimpleFunctionRegistryTests {
@Test
public void testFunctionCompositionExplicit() {
FunctionRegistration<Words> wordsRegistration = new FunctionRegistration<>(
new Words(), "words").type(FunctionType.of(Words.class));
new Words(), "words").type(Words.class);
FunctionRegistration<Reverse> reverseRegistration = new FunctionRegistration<>(
new Reverse(), "reverse").type(FunctionType.of(Reverse.class));
new Reverse(), "reverse").type(Reverse.class);
SimpleFunctionRegistry catalog = new SimpleFunctionRegistry(this.conversionService, this.messageConverter,
new JacksonMapper(new ObjectMapper()));
catalog.register(wordsRegistration);
@@ -363,10 +354,10 @@ public class SimpleFunctionRegistryTests {
public void testFunctionCompositionWithMessages() {
FunctionRegistration<UpperCaseMessage> upperCaseRegistration = new FunctionRegistration<>(
new UpperCaseMessage(), "uppercase")
.type(FunctionType.of(UpperCaseMessage.class));
.type(UpperCaseMessage.class);
FunctionRegistration<ReverseMessage> reverseRegistration = new FunctionRegistration<>(
new ReverseMessage(), "reverse")
.type(FunctionType.of(ReverseMessage.class));
.type(ReverseMessage.class);
SimpleFunctionRegistry catalog = new SimpleFunctionRegistry(this.conversionService, this.messageConverter,
new JacksonMapper(new ObjectMapper()));
catalog.register(upperCaseRegistration);
@@ -385,9 +376,9 @@ public class SimpleFunctionRegistryTests {
public void testFunctionCompositionMixedMessages() {
FunctionRegistration<UpperCaseMessage> upperCaseRegistration = new FunctionRegistration<>(
new UpperCaseMessage(), "uppercase")
.type(FunctionType.of(UpperCaseMessage.class));
.type(UpperCaseMessage.class);
FunctionRegistration<Reverse> reverseRegistration = new FunctionRegistration<>(
new Reverse(), "reverse").type(FunctionType.of(Reverse.class));
new Reverse(), "reverse").type(Reverse.class);
SimpleFunctionRegistry catalog = new SimpleFunctionRegistry(this.conversionService, this.messageConverter,
new JacksonMapper(new ObjectMapper()));
catalog.register(upperCaseRegistration);
@@ -405,7 +396,7 @@ public class SimpleFunctionRegistryTests {
@Test
public void testReactiveFunctionMessages() {
FunctionRegistration<ReactiveFunction> registration = new FunctionRegistration<>(new ReactiveFunction(), "reactive")
.type(FunctionType.of(ReactiveFunction.class));
.type(ReactiveFunction.class);
SimpleFunctionRegistry catalog = new SimpleFunctionRegistry(this.conversionService, this.messageConverter,
new JacksonMapper(new ObjectMapper()));
@@ -433,6 +424,7 @@ public class SimpleFunctionRegistryTests {
assertThat(result).isEqualTo("Jim Lahey");
}
@SuppressWarnings({ "rawtypes", "unchecked" })
@Test
public void lookup() {
SimpleFunctionRegistry functionRegistry = new SimpleFunctionRegistry(this.conversionService, this.messageConverter,
@@ -442,7 +434,7 @@ public class SimpleFunctionRegistryTests {
Function userFunction = uppercase();
FunctionRegistration functionRegistration = new FunctionRegistration(userFunction, "uppercase")
.type(FunctionType.from(String.class).to(String.class));
.type(FunctionTypeUtils.functionType(String.class, String.class));
functionRegistry.register(functionRegistration);
function = functionRegistry.lookup("uppercase");
@@ -450,20 +442,21 @@ public class SimpleFunctionRegistryTests {
}
@SuppressWarnings({ "rawtypes", "unchecked" })
@Test
public void lookupDefaultName() {
SimpleFunctionRegistry functionRegistry = new SimpleFunctionRegistry(this.conversionService, this.messageConverter,
new JacksonMapper(new ObjectMapper()));
Function userFunction = uppercase();
FunctionRegistration functionRegistration = new FunctionRegistration(userFunction, "uppercase")
.type(FunctionType.from(String.class).to(String.class));
.type(FunctionTypeUtils.functionType(String.class, String.class));
functionRegistry.register(functionRegistration);
FunctionInvocationWrapper function = functionRegistry.lookup("");
assertThat(function).isNotNull();
}
@SuppressWarnings("unchecked")
@SuppressWarnings({ "unchecked", "rawtypes" })
@Test
public void lookupWithCompositionFunctionAndConsumer() {
SimpleFunctionRegistry functionRegistry = new SimpleFunctionRegistry(this.conversionService, this.messageConverter,
@@ -471,7 +464,7 @@ public class SimpleFunctionRegistryTests {
Object userFunction = uppercase();
FunctionRegistration functionRegistration = new FunctionRegistration(userFunction, "uppercase")
.type(FunctionType.from(String.class).to(String.class));
.type(FunctionTypeUtils.functionType(String.class, String.class));
functionRegistry.register(functionRegistration);
userFunction = consumer();
@@ -484,6 +477,7 @@ public class SimpleFunctionRegistryTests {
functionWrapper.apply("123");
}
@SuppressWarnings({ "unchecked", "rawtypes" })
@Test
public void lookupWithReactiveConsumer() {
SimpleFunctionRegistry functionRegistry = new SimpleFunctionRegistry(this.conversionService, this.messageConverter,
@@ -500,12 +494,11 @@ public class SimpleFunctionRegistryTests {
functionWrapper.apply("123");
}
@SuppressWarnings({ "rawtypes", "unchecked" })
@Test
public void testHeaderEnricherFunction() {
FunctionRegistration<HeaderEnricherFunction> registration =
new FunctionRegistration<>(new HeaderEnricherFunction(), "headerEnricher")
.type(FunctionType.of(HeaderEnricherFunction.class));
.type(HeaderEnricherFunction.class);
SimpleFunctionRegistry catalog = new SimpleFunctionRegistry(this.conversionService, this.messageConverter,
new JacksonMapper(new ObjectMapper()));
catalog.register(registration);
@@ -519,7 +512,7 @@ public class SimpleFunctionRegistryTests {
@Test
public void testReactiveMonoSupplier() {
FunctionRegistration<ReactiveMonoGreeter> registration = new FunctionRegistration<>(new ReactiveMonoGreeter(),
"greeter").type(FunctionType.of(ReactiveMonoGreeter.class));
"greeter").type(ReactiveMonoGreeter.class);
SimpleFunctionRegistry catalog = new SimpleFunctionRegistry(this.conversionService, this.messageConverter,
new JacksonMapper(new ObjectMapper()));
catalog.register(registration);

20
spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogInitializerTests.java
View File

@@ -39,7 +39,7 @@ import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.cloud.function.context.FunctionCatalog;
import org.springframework.cloud.function.context.FunctionRegistration;
import org.springframework.cloud.function.context.FunctionType;
import org.springframework.cloud.function.context.catalog.FunctionTypeUtils;
import org.springframework.cloud.function.context.scan.TestFunction;
import org.springframework.context.ApplicationContextInitializer;
import org.springframework.context.annotation.Bean;
@@ -245,17 +245,18 @@ public class ContextFunctionCatalogInitializerTests {
private List<String> list = new ArrayList<>();
@Override
public void initialize(GenericApplicationContext context) {
context.registerBean("function", FunctionRegistration.class,
() -> new FunctionRegistration<>(function()).type(
FunctionType.from(Person.class).to(Person.class).getType()));
() -> new FunctionRegistration<>(function()).type(FunctionTypeUtils.functionType(Person.class, Person.class)));
context.registerBean("supplier", FunctionRegistration.class,
() -> new FunctionRegistration<>(supplier())
.type(FunctionType.supplier(String.class).getType()));
.type(FunctionTypeUtils.supplierType(String.class)));
context.registerBean("consumer", FunctionRegistration.class,
() -> new FunctionRegistration<>(consumer())
.type(FunctionType.consumer(String.class).getType()));
.type(FunctionTypeUtils.consumerType(String.class)));
context.registerBean(SimpleConfiguration.class, () -> this);
}
@@ -296,8 +297,7 @@ public class ContextFunctionCatalogInitializerTests {
@Override
public void initialize(GenericApplicationContext context) {
context.registerBean("function", FunctionRegistration.class,
() -> new FunctionRegistration<>(function()).type(
FunctionType.from(String.class).to(String.class).getType()));
() -> new FunctionRegistration<>(function()).type(FunctionTypeUtils.functionType(String.class, String.class)));
context.registerBean(PropertiesConfiguration.class, () -> this);
}
@@ -317,8 +317,7 @@ public class ContextFunctionCatalogInitializerTests {
@Override
public void initialize(GenericApplicationContext context) {
context.registerBean("function", FunctionRegistration.class,
() -> new FunctionRegistration<>(function()).type(
FunctionType.from(String.class).to(String.class).getType()));
() -> new FunctionRegistration<>(function()).type(FunctionTypeUtils.functionType(String.class, String.class)));
context.registerBean(ValueConfiguration.class, () -> this);
}
@@ -355,8 +354,7 @@ public class ContextFunctionCatalogInitializerTests {
context.registerBean(String.class, () -> value());
context.registerBean("foos", FunctionRegistration.class,
() -> new FunctionRegistration<>(foos(context.getBean(String.class)))
.type(FunctionType.from(String.class).to(Foo.class)
.getType()));
.type(FunctionTypeUtils.functionType(String.class, Foo.class)));
}
@Bean