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

GH-1266 Improve and clean up SimpleFunctionRegistry

Browse Source 
Resolves #1266
This commit is contained in:
Oleg Zhurakousky
2025-04-16 19:50:26 +02:00
parent d20aa3d0bd
commit f494962000
4 changed files with 28 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -70,7 +70,7 @@ public class CloudEventsFunctionInvocationHelper implements FunctionInvocationHe
}
@Override
public boolean isRetainOuputAsMessage(Message<?> message) {
public boolean isRetainOutputAsMessage(Message<?> message) {
return message.getHeaders().containsKey(MessageUtils.TARGET_PROTOCOL) || (message.getHeaders().containsKey(MessageUtils.MESSAGE_TYPE)
&& message.getHeaders().get(MessageUtils.MESSAGE_TYPE).equals(CloudEventMessageUtils.CLOUDEVENT_VALUE));
}

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

@@ -21,6 +21,7 @@ import java.lang.reflect.Type;
import java.lang.reflect.TypeVariable;
import java.lang.reflect.WildcardType;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -184,11 +185,10 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
boolean isFunctionDefinitionEligible(String functionDefinition) {
if (this.functionProperties != null) {
for (String definition : this.functionProperties.getIneligibleDefinitions()) {
if (functionDefinition.contains(definition)) {
return false;
}
}
this.functionProperties.getIneligibleDefinitions().contains(functionDefinition);
boolean matchFoundInBoth = !Collections.disjoint(Arrays.asList(functionDefinition.split("\\|")),
this.functionProperties.getIneligibleDefinitions());
return !matchFoundInBoth;
}
return true;
}
@@ -563,9 +563,6 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
if (FunctionTypeUtils.isPublisher(type) || FunctionTypeUtils.isMessage(type) || FunctionTypeUtils.isTypeCollection(type)) {
type = FunctionTypeUtils.getGenericType(type);
}
if (FunctionTypeUtils.isMessage(type)) {
type = FunctionTypeUtils.getGenericType(type);
}
return type;
}
@@ -808,18 +805,11 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
*/
private Object enrichInvocationResultIfNecessary(Object input, Object result) {
if (result != null && !(result instanceof Publisher) && input instanceof Message) {
if (result instanceof Message) {
if (functionInvocationHelper != null && CloudEventMessageUtils.isCloudEvent(((Message) input))) {
result = functionInvocationHelper.postProcessResult(result, (Message) input);
}
if (functionInvocationHelper != null && CloudEventMessageUtils.isCloudEvent(((Message) input))) {
result = functionInvocationHelper.postProcessResult(result, (Message) input);
}
else {
if (functionInvocationHelper != null && CloudEventMessageUtils.isCloudEvent(((Message) input))) {
result = functionInvocationHelper.postProcessResult(result, (Message) input);
}
else if (!FunctionTypeUtils.isCollectionOfMessage(this.outputType)) {
result = MessageBuilder.withPayload(result).copyHeaders(this.sanitizeHeaders(((Message) input).getHeaders())).build();
}
if (!FunctionTypeUtils.isCollectionOfMessage(this.outputType)) {
result = MessageBuilder.withPayload(result).copyHeaders(this.sanitizeHeaders(((Message) input).getHeaders())).build();
}
}
return result;
@@ -920,21 +910,21 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
@SuppressWarnings("unchecked")
private Object invokeFunction(Object convertedInput) {
Object result;
if (!this.isTypePublisher(this.inputType) && convertedInput instanceof Publisher) {
result = convertedInput instanceof Mono
? Mono.from((Publisher) convertedInput).map(value -> this.invokeFunctionAndEnrichResultIfNecessary(value))
if (!this.isTypePublisher(this.inputType) && convertedInput instanceof Publisher publisherInput) {
result = publisherInput instanceof Mono
? Mono.from(publisherInput).map(value -> this.invokeFunctionAndEnrichResultIfNecessary(value))
.doOnError(ex -> logger.error("Failed to invoke function '" + this.functionDefinition + "'", (Throwable) ex))
: Flux.from((Publisher) convertedInput).map(value -> this.invokeFunctionAndEnrichResultIfNecessary(value))
: Flux.from(publisherInput).map(value -> this.invokeFunctionAndEnrichResultIfNecessary(value))
.doOnError(ex -> logger.error("Failed to invoke function '" + this.functionDefinition + "'", (Throwable) ex));
}
else {
result = this.invokeFunctionAndEnrichResultIfNecessary(convertedInput);
if (result instanceof Flux) {
result = ((Flux) result).doOnError(ex -> logger.error("Failed to invoke function '"
if (result instanceof Flux flux) {
result = flux.doOnError(ex -> logger.error("Failed to invoke function '"
+ this.functionDefinition + "'", (Throwable) ex));
}
else if (result instanceof Mono) {
result = ((Mono) result).doOnError(ex -> logger.error("Failed to invoke function '"
else if (result instanceof Mono mono) {
result = mono.doOnError(ex -> logger.error("Failed to invoke function '"
+ this.functionDefinition + "'", (Throwable) ex));
}
}
@@ -989,8 +979,8 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
result = this.postProcessFunction((Publisher) result, firstInputMessage);
}
return value instanceof OriginalMessageHolder
? this.enrichInvocationResultIfNecessary(((OriginalMessageHolder) value).getOriginalMessage(), result)
return value instanceof OriginalMessageHolder originalMessageHolder
? this.enrichInvocationResultIfNecessary((originalMessageHolder).getOriginalMessage(), result)
: result;
}
@@ -1035,8 +1025,8 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
private Object invokeConsumer(Object convertedInput) {
Object result = null;
if (this.isTypePublisher(this.inputType)) {
if (convertedInput instanceof Flux) {
result = ((Flux) convertedInput)
if (convertedInput instanceof Flux fluxInput) {
result = fluxInput
.transform(flux -> {
flux = Flux.from((Publisher) flux).map(v -> this.extractValueFromOriginalValueHolderIfNecessary(v));
((Consumer) this.target).accept(flux);
@@ -1052,12 +1042,12 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
}).then();
}
}
else if (convertedInput instanceof Publisher) {
else if (convertedInput instanceof Publisher publisherInput) {
result = convertedInput instanceof Mono
? Mono.from((Publisher) convertedInput)
? Mono.from(publisherInput)
.map(v -> this.extractValueFromOriginalValueHolderIfNecessary(v))
.doOnNext((Consumer) this.target).then()
: Flux.from((Publisher) convertedInput)
: Flux.from(publisherInput)
.map(v -> this.extractValueFromOriginalValueHolderIfNecessary(v))
.doOnNext((Consumer) this.target).then();
}
@@ -1185,18 +1175,12 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
}
private boolean isExtractPayload(Message<?> message, Type type) {
if (this.propagateInputHeaders) {
return false;
}
if (this.isRoutingFunction()) {
if (this.propagateInputHeaders || this.isRoutingFunction() || FunctionTypeUtils.isMessage(type)) {
return false;
}
if (FunctionTypeUtils.isCollectionOfMessage(type)) {
return true;
}
if (FunctionTypeUtils.isMessage(type)) {
return false;
}
Object payload = message.getPayload();
if ((payload instanceof byte[])) {
@@ -1291,7 +1275,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
* case that requires it since it may contain forwarding url
*/
private boolean containsRetainMessageSignalInHeaders(Message message) {
if (functionInvocationHelper != null && functionInvocationHelper.isRetainOuputAsMessage(message)) {
if (functionInvocationHelper != null && functionInvocationHelper.isRetainOutputAsMessage(message)) {
return true;
}
else {
@@ -1348,7 +1332,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
*
*/
private Type extractActualValueTypeIfNecessary(Type type) {
if (type instanceof ParameterizedType && (FunctionTypeUtils.isPublisher(type) || FunctionTypeUtils.isMessage(type))) {
if (type instanceof ParameterizedType && (FunctionTypeUtils.isPublisher(type) || FunctionTypeUtils.isMessage(type))) {
return FunctionTypeUtils.getGenericType(type);
}
return type;

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

@@ -87,12 +87,6 @@ public abstract class FunctionContextUtils {
return param;
}
@Deprecated(since = "4.2.3", forRemoval = true)
public static Class<?>[] getParamTypesFromBeanDefinitionFactory(Class<?> factory,
AbstractBeanDefinition definition) {
return getParamTypesFromBeanDefinitionFactory(factory, definition, null);
}
public static Class<?>[] getParamTypesFromBeanDefinitionFactory(Class<?> factory,
AbstractBeanDefinition definition, String methodName) {
if (definition instanceof RootBeanDefinition) {