diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/cloudevent/CloudEventsFunctionInvocationHelper.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/cloudevent/CloudEventsFunctionInvocationHelper.java index 9d1b6fa2b..136de7949 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/cloudevent/CloudEventsFunctionInvocationHelper.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/cloudevent/CloudEventsFunctionInvocationHelper.java @@ -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)); } diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java index 490ac01c3..39cea3a0e 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java @@ -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; diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/FunctionContextUtils.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/FunctionContextUtils.java index 27c0c6239..dacf7b74a 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/FunctionContextUtils.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/FunctionContextUtils.java @@ -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) { diff --git a/spring-cloud-function-core/src/main/java/org/springframework/cloud/function/core/FunctionInvocationHelper.java b/spring-cloud-function-core/src/main/java/org/springframework/cloud/function/core/FunctionInvocationHelper.java index 5c72fb92b..0698fdbff 100644 --- a/spring-cloud-function-core/src/main/java/org/springframework/cloud/function/core/FunctionInvocationHelper.java +++ b/spring-cloud-function-core/src/main/java/org/springframework/cloud/function/core/FunctionInvocationHelper.java @@ -31,14 +31,6 @@ public interface FunctionInvocationHelper { return true; } - /** - * @deprecated Use {@link #isRetainOutputAsMessage(I)} - */ - @Deprecated - default boolean isRetainOuputAsMessage(I input) { - return isRetainOutputAsMessage(input); - }; - default I preProcessInput(I input, Object inputConverter) { return input; }