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

Revert "Add a new strategy for header enrichment"

Browse Source 
This reverts commit 1925b490dc.
This commit is contained in:
Oleg Zhurakousky
2020-11-30 15:23:42 +01:00
parent 1925b490dc
commit dd6c09a4cb
7 changed files with 189 additions and 391 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/cloudevent/CloudEventOutputMessageHeaderEnricher.java
View File

@@ -1,83 +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.cloudevent;
import org.springframework.beans.BeansException;
import org.springframework.cloud.function.context.message.OutputMessageHeaderEnricher;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.Ordered;
import org.springframework.core.env.Environment;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.ClassUtils;
import org.springframework.util.StringUtils;
/**
* @author Dave Syer
*
*/
public class CloudEventOutputMessageHeaderEnricher
implements OutputMessageHeaderEnricher, ApplicationContextAware, Ordered {
private ApplicationContext applicationContext;
private CloudEventAttributesProvider cloudEventAttributesProvider;
private static final String CLOUD_EVENT_TYPE_NAME = "io.cloudevents.api.CloudEvent";
private static Class<?> CLOUD_EVENT_TYPE = ClassUtils.isPresent(CLOUD_EVENT_TYPE_NAME, null)
? ClassUtils.resolveClassName(CLOUD_EVENT_TYPE_NAME, null) : null;
@Override
public int getOrder() {
return 0;
}
@Override
public Message<?> enrich(Message<?> output) {
Object invocationResult = output.getPayload();
if (CLOUD_EVENT_TYPE != null && CLOUD_EVENT_TYPE.isAssignableFrom(invocationResult.getClass())) {
// User is sending us an actual CloudEvent, so no need to guess the attributes
return output;
}
CloudEventAttributes generatedCeHeaders = CloudEventMessageUtils.generateAttributes(output,
invocationResult.getClass().getName(), getApplicationName());
CloudEventAttributes attributes = new CloudEventAttributes(generatedCeHeaders,
CloudEventMessageUtils.determinePrefixToUse(output.getHeaders()));
if (cloudEventAttributesProvider != null) {
// Global defaults can easily be changed by injecting one of these
cloudEventAttributesProvider.generateDefaultCloudEventHeaders(attributes);
}
return MessageBuilder.withPayload(invocationResult).copyHeaders(attributes).build();
}
private String getApplicationName() {
Environment environment = this.applicationContext.getEnvironment();
String name = environment.getProperty("spring.application.name");
return "http://spring.io/"
+ (StringUtils.hasText(name) ? name : "application-" + this.applicationContext.getId());
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = applicationContext;
if (applicationContext.getBeanNamesForType(CloudEventAttributesProvider.class).length > 0) {
this.cloudEventAttributesProvider = applicationContext.getBean(CloudEventAttributesProvider.class);
}
}
}

44
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistry.java
View File

@@ -20,10 +20,10 @@ import java.lang.reflect.Method;
import java.lang.reflect.Type; import java.lang.reflect.Type;
import java.util.Arrays; import java.util.Arrays;
import java.util.Set; import java.util.Set;
import java.util.function.BiFunction;
import java.util.function.Consumer; import java.util.function.Consumer;
import java.util.function.Function; import java.util.function.Function;
import java.util.function.Supplier; import java.util.function.Supplier;
import java.util.stream.Collectors;
import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation; import org.aopalliance.intercept.MethodInvocation;
@@ -32,17 +32,21 @@ import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.BeansException; import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.annotation.BeanFactoryAnnotationUtils; import org.springframework.beans.factory.annotation.BeanFactoryAnnotationUtils;
import org.springframework.cloud.function.cloudevent.CloudEventAttributes;
import org.springframework.cloud.function.cloudevent.CloudEventAttributesProvider;
import org.springframework.cloud.function.cloudevent.CloudEventMessageUtils;
import org.springframework.cloud.function.context.FunctionProperties; import org.springframework.cloud.function.context.FunctionProperties;
import org.springframework.cloud.function.context.FunctionRegistration; import org.springframework.cloud.function.context.FunctionRegistration;
import org.springframework.cloud.function.context.FunctionRegistry; import org.springframework.cloud.function.context.FunctionRegistry;
import org.springframework.cloud.function.context.message.CompositeOutputMessageHeaderEnricher;
import org.springframework.cloud.function.context.message.OutputMessageHeaderEnricher;
import org.springframework.cloud.function.json.JsonMapper; import org.springframework.cloud.function.json.JsonMapper;
import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware; import org.springframework.context.ApplicationContextAware;
import org.springframework.context.support.GenericApplicationContext; import org.springframework.context.support.GenericApplicationContext;
import org.springframework.core.convert.ConversionService; import org.springframework.core.convert.ConversionService;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.messaging.Message;
import org.springframework.messaging.converter.CompositeMessageConverter; import org.springframework.messaging.converter.CompositeMessageConverter;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.StringUtils; import org.springframework.util.StringUtils;
/** /**
@@ -55,7 +59,7 @@ public class BeanFactoryAwareFunctionRegistry extends SimpleFunctionRegistry imp
private GenericApplicationContext applicationContext; private GenericApplicationContext applicationContext;
private OutputMessageHeaderEnricher enricher; private CloudEventAttributesProvider cloudEventAtttributesProvider;
public BeanFactoryAwareFunctionRegistry(ConversionService conversionService, public BeanFactoryAwareFunctionRegistry(ConversionService conversionService,
CompositeMessageConverter messageConverter, JsonMapper jsonMapper) { CompositeMessageConverter messageConverter, JsonMapper jsonMapper) {
@@ -65,9 +69,8 @@ public class BeanFactoryAwareFunctionRegistry extends SimpleFunctionRegistry imp
@Override @Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = (GenericApplicationContext) applicationContext; this.applicationContext = (GenericApplicationContext) applicationContext;
if (applicationContext.getBeanNamesForType(OutputMessageHeaderEnricher.class).length > 0) { if (applicationContext.getBeanNamesForType(CloudEventAttributesProvider.class).length > 0) {
this.enricher = new CompositeOutputMessageHeaderEnricher(applicationContext this.cloudEventAtttributesProvider = applicationContext.getBean(CloudEventAttributesProvider.class);
.getBeanProvider(OutputMessageHeaderEnricher.class).orderedStream().collect(Collectors.toList()));
} }
} }
@@ -161,13 +164,36 @@ public class BeanFactoryAwareFunctionRegistry extends SimpleFunctionRegistry imp
function = super.doLookup(type, functionDefinition, expectedOutputMimeTypes); function = super.doLookup(type, functionDefinition, expectedOutputMimeTypes);
} }
if (function != null && this.enricher != null) { if (function != null) {
function.setOutputMessageHeaderEnricher(this.enricher); BiFunction<Message<?>, Object, Message<?>> invocationResultHeaderEnricher = new BiFunction<Message<?>, Object, Message<?>>() {
@Override
public Message<?> apply(Message<?> inputMessage, Object invocationResult) {
// TODO: Factor it out! Cloud Events specific code
CloudEventAttributes generatedCeHeaders = CloudEventMessageUtils.generateAttributes(inputMessage,
invocationResult.getClass().getName(), getApplicationName());
CloudEventAttributes attributes = new CloudEventAttributes(generatedCeHeaders,
CloudEventMessageUtils.determinePrefixToUse(inputMessage.getHeaders()));
if (cloudEventAtttributesProvider != null) {
cloudEventAtttributesProvider.generateDefaultCloudEventHeaders(attributes);
}
Message message = MessageBuilder.withPayload(invocationResult).copyHeaders(attributes).build();
return message;
}
};
function.setOutputMessageHeaderEnricher(invocationResultHeaderEnricher);
} }
return (T) function; return (T) function;
} }
private String getApplicationName() {
ConfigurableEnvironment environment = this.applicationContext.getEnvironment();
String name = environment.getProperty("spring.application.name");
return "http://spring.io/"
+ (StringUtils.hasText(name) ? name : "application-" + this.applicationContext.getId());
}
private Object discoverFunctionInBeanFactory(String functionName) { private Object discoverFunctionInBeanFactory(String functionName) {
Object functionCandidate = null; Object functionCandidate = null;
if (this.applicationContext.containsBean(functionName)) { if (this.applicationContext.containsBean(functionName)) {

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

@@ -31,6 +31,7 @@ import java.util.Map;
import java.util.Optional; import java.util.Optional;
import java.util.Set; import java.util.Set;
import java.util.TreeSet; import java.util.TreeSet;
import java.util.function.BiFunction;
import java.util.function.Consumer; import java.util.function.Consumer;
import java.util.function.Function; import java.util.function.Function;
import java.util.function.Supplier; import java.util.function.Supplier;
@@ -52,7 +53,6 @@ import org.springframework.cloud.function.context.FunctionProperties;
import org.springframework.cloud.function.context.FunctionRegistration; import org.springframework.cloud.function.context.FunctionRegistration;
import org.springframework.cloud.function.context.FunctionRegistry; import org.springframework.cloud.function.context.FunctionRegistry;
import org.springframework.cloud.function.context.config.RoutingFunction; import org.springframework.cloud.function.context.config.RoutingFunction;
import org.springframework.cloud.function.context.message.OutputMessageHeaderEnricher;
import org.springframework.cloud.function.json.JsonMapper; import org.springframework.cloud.function.json.JsonMapper;
import org.springframework.core.ResolvableType; import org.springframework.core.ResolvableType;
import org.springframework.core.convert.ConversionService; import org.springframework.core.convert.ConversionService;
@@ -69,21 +69,20 @@ import org.springframework.util.ObjectUtils;
import org.springframework.util.ReflectionUtils; import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils; import org.springframework.util.StringUtils;
/** /**
* Implementation of {@link FunctionCatalog} and {@link FunctionRegistry} which does not * Implementation of {@link FunctionCatalog} and {@link FunctionRegistry} which
* depend on Spring's {@link BeanFactory}. Each function must be registered with it * does not depend on Spring's {@link BeanFactory}.
* explicitly to benefit from features such as type conversion, composition, POJO etc. * Each function must be registered with it explicitly to benefit from features
* such as type conversion, composition, POJO etc.
* *
* @author Oleg Zhurakousky * @author Oleg Zhurakousky
* *
*/ */
public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspector { public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspector {
protected Log logger = LogFactory.getLog(this.getClass()); protected Log logger = LogFactory.getLog(this.getClass());
/* /*
* - do we care about FunctionRegistration after it's been registered? What additional * - do we care about FunctionRegistration after it's been registered? What additional value does it bring?
* value does it bring?
* *
*/ */
@@ -102,8 +101,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
@Autowired(required = false) @Autowired(required = false)
private FunctionAroundWrapper functionAroundWrapper; private FunctionAroundWrapper functionAroundWrapper;
public SimpleFunctionRegistry(ConversionService conversionService, CompositeMessageConverter messageConverter, public SimpleFunctionRegistry(ConversionService conversionService, CompositeMessageConverter messageConverter, JsonMapper jsonMapper) {
JsonMapper jsonMapper) {
Assert.notNull(messageConverter, "'messageConverter' must not be null"); Assert.notNull(messageConverter, "'messageConverter' must not be null");
Assert.notNull(jsonMapper, "'jsonMapper' must not be null"); Assert.notNull(jsonMapper, "'jsonMapper' must not be null");
this.conversionService = conversionService; this.conversionService = conversionService;
@@ -145,7 +143,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
this.functionRegistrations.add(registration); this.functionRegistrations.add(registration);
} }
// ----- //-----
@Override @Override
public Set<String> getNames(Class<?> type) { public Set<String> getNames(Class<?> type) {
@@ -175,7 +173,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
function = this.compose(type, functionDefinition); function = this.compose(type, functionDefinition);
} }
if (function != null && !ObjectUtils.isEmpty(expectedOutputMimeTypes)) { if (function != null && !ObjectUtils.isEmpty(expectedOutputMimeTypes)) {
function.expectedOutputContentType = expectedOutputMimeTypes; function.expectedOutputContentType = expectedOutputMimeTypes;
} }
else if (logger.isDebugEnabled()) { else if (logger.isDebugEnabled()) {
@@ -188,19 +186,23 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
} }
/** /**
* This method will make sure that if there is only one function in catalog it can be * This method will make sure that if there is only one function in catalog
* looked up by any name or no name. It does so by attempting to determine the default * it can be looked up by any name or no name.
* function name (the only function in catalog) and checking if it matches the * It does so by attempting to determine the default function name
* provided name replacing it if it does not. * (the only function in catalog) and checking if it matches the provided name
* replacing it if it does not.
*/ */
String normalizeFunctionDefinition(String functionDefinition) { String normalizeFunctionDefinition(String functionDefinition) {
functionDefinition = StringUtils.hasText(functionDefinition) ? functionDefinition.replaceAll(",", "|") functionDefinition = StringUtils.hasText(functionDefinition)
? functionDefinition.replaceAll(",", "|")
: System.getProperty(FunctionProperties.FUNCTION_DEFINITION, ""); : System.getProperty(FunctionProperties.FUNCTION_DEFINITION, "");
if (!this.getNames(null).contains(functionDefinition)) { if (!this.getNames(null).contains(functionDefinition)) {
List<String> eligibleFunction = this.getNames(null).stream() List<String> eligibleFunction = this.getNames(null).stream()
.filter(name -> !RoutingFunction.FUNCTION_NAME.equals(name)).collect(Collectors.toList()); .filter(name -> !RoutingFunction.FUNCTION_NAME.equals(name))
if (eligibleFunction.size() == 1 && !eligibleFunction.get(0).equals(functionDefinition) .collect(Collectors.toList());
if (eligibleFunction.size() == 1
&& !eligibleFunction.get(0).equals(functionDefinition)
&& !functionDefinition.contains("|")) { && !functionDefinition.contains("|")) {
functionDefinition = eligibleFunction.get(0); functionDefinition = eligibleFunction.get(0);
} }
@@ -209,8 +211,9 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
} }
/** /**
* This is primarily to support spring-cloud-sleauth. There is no current use cases in * This is primarily to support spring-cloud-sleauth.
* functions where it is used. The approach may change in the future. * There is no current use cases in functions where it is used.
* The approach may change in the future.
*/ */
private FunctionInvocationWrapper wrapInAroundAviceIfNecessary(FunctionInvocationWrapper function) { private FunctionInvocationWrapper wrapInAroundAviceIfNecessary(FunctionInvocationWrapper function) {
FunctionInvocationWrapper wrappedFunction = function; FunctionInvocationWrapper wrappedFunction = function;
@@ -231,9 +234,12 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
*/ */
private FunctionInvocationWrapper findFunctionInFunctionRegistrations(String functionName) { private FunctionInvocationWrapper findFunctionInFunctionRegistrations(String functionName) {
FunctionRegistration<?> functionRegistration = this.functionRegistrations.stream() FunctionRegistration<?> functionRegistration = this.functionRegistrations.stream()
.filter(fr -> fr.getNames().contains(functionName)).findFirst().orElseGet(() -> null); .filter(fr -> fr.getNames().contains(functionName))
return functionRegistration != null ? this.invocationWrapperInstance(functionName, .findFirst()
functionRegistration.getTarget(), functionRegistration.getType().getType()) : null; .orElseGet(() -> null);
return functionRegistration != null
? this.invocationWrapperInstance(functionName, functionRegistration.getTarget(), functionRegistration.getType().getType())
: null;
} }
@@ -241,8 +247,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
* *
*/ */
private FunctionInvocationWrapper compose(Class<?> type, String functionDefinition) { private FunctionInvocationWrapper compose(Class<?> type, String functionDefinition) {
String[] functionNames = StringUtils.delimitedListToStringArray(functionDefinition.replaceAll(",", "|").trim(), String[] functionNames = StringUtils.delimitedListToStringArray(functionDefinition.replaceAll(",", "|").trim(), "|");
"|");
FunctionInvocationWrapper composedFunction = null; FunctionInvocationWrapper composedFunction = null;
for (String functionName : functionNames) { for (String functionName : functionNames) {
@@ -255,10 +260,9 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
composedFunction = function; composedFunction = function;
} }
else { else {
FunctionInvocationWrapper andThenFunction = invocationWrapperInstance(functionName, FunctionInvocationWrapper andThenFunction =
function.getTarget(), function.inputType, function.outputType); invocationWrapperInstance(functionName, function.getTarget(), function.inputType, function.outputType);
composedFunction = (FunctionInvocationWrapper) composedFunction composedFunction = (FunctionInvocationWrapper) composedFunction.andThen((Function<Object, Object>) andThenFunction);
.andThen((Function<Object, Object>) andThenFunction);
} }
this.wrappedFunctionDefinitions.put(composedFunction.functionDefinition, composedFunction); this.wrappedFunctionDefinitions.put(composedFunction.functionDefinition, composedFunction);
} }
@@ -272,16 +276,14 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
/* /*
* *
*/ */
private FunctionInvocationWrapper invocationWrapperInstance(String functionDefinition, Object target, private FunctionInvocationWrapper invocationWrapperInstance(String functionDefinition, Object target, Type inputType, Type outputType) {
Type inputType, Type outputType) {
return new FunctionInvocationWrapper(functionDefinition, target, inputType, outputType); return new FunctionInvocationWrapper(functionDefinition, target, inputType, outputType);
} }
/* /*
* *
*/ */
private FunctionInvocationWrapper invocationWrapperInstance(String functionDefinition, Object target, private FunctionInvocationWrapper invocationWrapperInstance(String functionDefinition, Object target, Type functionType) {
Type functionType) {
return invocationWrapperInstance(functionDefinition, target, return invocationWrapperInstance(functionDefinition, target,
FunctionTypeUtils.isSupplier(functionType) ? null : FunctionTypeUtils.getInputType(functionType), FunctionTypeUtils.isSupplier(functionType) ? null : FunctionTypeUtils.getInputType(functionType),
FunctionTypeUtils.getOutputType(functionType)); FunctionTypeUtils.getOutputType(functionType));
@@ -291,8 +293,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
* *
*/ */
@SuppressWarnings("rawtypes") @SuppressWarnings("rawtypes")
public class FunctionInvocationWrapper public class FunctionInvocationWrapper implements Function<Object, Object>, Consumer<Object>, Supplier<Object>, Runnable {
implements Function<Object, Object>, Consumer<Object>, Supplier<Object>, Runnable {
private final Object target; private final Object target;
@@ -313,17 +314,17 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
private boolean skipOutputConversion; private boolean skipOutputConversion;
/* /*
* This is primarily to support Stream's ability to access un-converted payload * This is primarily to support Stream's ability to access
* (e.g., to evaluate expression on some attribute of a payload) It does not have * un-converted payload (e.g., to evaluate expression on some attribute of a payload)
* a setter/getter and can only be set via reflection. It is not intended to * It does not have a setter/getter and can only be set via reflection.
* remain here and will be removed as soon as particular elements of stream will * It is not intended to remain here and will be removed as soon as particular elements
* be refactored to address this. * of stream will be refactored to address this.
*/ */
private Function<Object, Message> enhancer; private Function<Object, Message> enhancer;
private OutputMessageHeaderEnricher outputMessageHeaderEnricher; private BiFunction<Message<?>, Object, Message<?>> outputMessageHeaderEnricher;
void setOutputMessageHeaderEnricher(OutputMessageHeaderEnricher outputMessageHeaderEnricher) { void setOutputMessageHeaderEnricher(BiFunction<Message<?>, Object, Message<?>> outputMessageHeaderEnricher) {
this.outputMessageHeaderEnricher = outputMessageHeaderEnricher; this.outputMessageHeaderEnricher = outputMessageHeaderEnricher;
} }
@@ -335,7 +336,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
this.message = this.inputType != null && FunctionTypeUtils.isMessage(this.inputType); this.message = this.inputType != null && FunctionTypeUtils.isMessage(this.inputType);
} }
FunctionInvocationWrapper(String functionDefinition, Object target, Type inputType, Type outputType) { FunctionInvocationWrapper(String functionDefinition, Object target, Type inputType, Type outputType) {
this.target = target; this.target = target;
this.inputType = this.normalizeType(inputType); this.inputType = this.normalizeType(inputType);
this.outputType = this.normalizeType(outputType); this.outputType = this.normalizeType(outputType);
@@ -345,16 +346,14 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
public void setSkipInputConversion(boolean skipInputConversion) { public void setSkipInputConversion(boolean skipInputConversion) {
if (logger.isDebugEnabled() && skipInputConversion) { if (logger.isDebugEnabled() && skipInputConversion) {
logger.debug( logger.debug("'skipInputConversion' was explicitely set to true. No input conversion will be attempted");
"'skipInputConversion' was explicitely set to true. No input conversion will be attempted");
} }
this.skipInputConversion = skipInputConversion; this.skipInputConversion = skipInputConversion;
} }
public void setSkipOutputConversion(boolean skipOutputConversion) { public void setSkipOutputConversion(boolean skipOutputConversion) {
if (logger.isDebugEnabled() && skipOutputConversion) { if (logger.isDebugEnabled() && skipOutputConversion) {
logger.debug( logger.debug("'skipOutputConversion' was explicitely set to true. No output conversion will be attempted");
"'skipOutputConversion' was explicitely set to true. No output conversion will be attempted");
} }
this.skipOutputConversion = skipOutputConversion; this.skipOutputConversion = skipOutputConversion;
} }
@@ -372,27 +371,23 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
} }
/** /**
* Return the actual {@link Type} of the item of the provided type. This method is * Return the actual {@link Type} of the item of the provided type.
* context specific and is not a general purpose utility method. The context is * This method is context specific and is not a general purpose utility method. The context is that the provided
* that the provided {@link Type} may represent the input/output of a function * {@link Type} may represent the input/output of a function where such type could be wrapped in
* where such type could be wrapped in {@link Message}, {@link Flux} or * {@link Message}, {@link Flux} or {@link Mono}, so this method returns generic value of such type or itself if not wrapped.
* {@link Mono}, so this method returns generic value of such type or itself if * @param type typically input or output Type of the function (see {@link #getInputType()} or {@link #getOutputType()}.
* not wrapped.
* @param type typically input or output Type of the function (see
* {@link #getInputType()} or {@link #getOutputType()}.
* @return the type of the item if wrapped otherwise the provided type. * @return the type of the item if wrapped otherwise the provided type.
*/ */
public Type getItemType(Type type) { public Type getItemType(Type type) {
if (FunctionTypeUtils.isPublisher(type) || FunctionTypeUtils.isMessage(type) if (FunctionTypeUtils.isPublisher(type) || FunctionTypeUtils.isMessage(type) || FunctionTypeUtils.isTypeCollection(type)) {
|| FunctionTypeUtils.isTypeCollection(type)) {
type = FunctionTypeUtils.getGenericType(type); type = FunctionTypeUtils.getGenericType(type);
} }
return type; return type;
} }
/** /**
* Use individual {@link #getInputType()}, {@link #getOutputType()} and their * Use individual {@link #getInputType()}, {@link #getOutputType()} and their variants as well as
* variants as well as other supporting operations instead. * other supporting operations instead.
* @deprecated since 3.1 * @deprecated since 3.1
*/ */
@Deprecated @Deprecated
@@ -425,7 +420,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
*/ */
@Override @Override
public Object apply(Object input) { public Object apply(Object input) {
if (logger.isDebugEnabled() && !(input instanceof Publisher)) { if (logger.isDebugEnabled() && !(input instanceof Publisher)) {
logger.debug("Invoking function " + this); logger.debug("Invoking function " + this);
} }
Object result = this.doApply(input); Object result = this.doApply(input);
@@ -481,6 +476,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
return FunctionTypeUtils.isMessage(this.outputType); return FunctionTypeUtils.isMessage(this.outputType);
} }
public boolean isRoutingFunction() { public boolean isRoutingFunction() {
return this.target instanceof RoutingFunction; return this.target instanceof RoutingFunction;
} }
@@ -491,14 +487,12 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
@Override @Override
public <V> Function<Object, V> andThen(Function<? super Object, ? extends V> after) { public <V> Function<Object, V> andThen(Function<? super Object, ? extends V> after) {
Assert.isTrue(after instanceof FunctionInvocationWrapper, Assert.isTrue(after instanceof FunctionInvocationWrapper, "Composed function must be an instanceof FunctionInvocationWrapper.");
"Composed function must be an instanceof FunctionInvocationWrapper.");
if (FunctionTypeUtils.isMultipleArgumentType(this.inputType) if (FunctionTypeUtils.isMultipleArgumentType(this.inputType)
|| FunctionTypeUtils.isMultipleArgumentType(this.outputType) || FunctionTypeUtils.isMultipleArgumentType(this.outputType)
|| FunctionTypeUtils.isMultipleArgumentType(((FunctionInvocationWrapper) after).inputType) || FunctionTypeUtils.isMultipleArgumentType(((FunctionInvocationWrapper) after).inputType)
|| FunctionTypeUtils.isMultipleArgumentType(((FunctionInvocationWrapper) after).outputType)) { || FunctionTypeUtils.isMultipleArgumentType(((FunctionInvocationWrapper) after).outputType)) {
throw new UnsupportedOperationException( throw new UnsupportedOperationException("Composition of functions with multiple arguments is not supported at the moment");
"Composition of functions with multiple arguments is not supported at the moment");
} }
Function rawComposedFunction = v -> ((FunctionInvocationWrapper) after).doApply(doApply(v)); Function rawComposedFunction = v -> ((FunctionInvocationWrapper) after).doApply(doApply(v));
@@ -507,39 +501,35 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
Type composedFunctionType; Type composedFunctionType;
if (afterWrapper.outputType == null) { if (afterWrapper.outputType == null) {
composedFunctionType = ResolvableType.forClassWithGenerics(Consumer.class, composedFunctionType = ResolvableType.forClassWithGenerics(Consumer.class, this.inputType == null
this.inputType == null ? null : ResolvableType.forType(this.inputType)).getType(); ? null
: ResolvableType.forType(this.inputType)).getType();
} }
else if (this.inputType == null && afterWrapper.outputType != null) { else if (this.inputType == null && afterWrapper.outputType != null) {
ResolvableType composedOutputType; ResolvableType composedOutputType;
if (FunctionTypeUtils.isFlux(this.outputType)) { if (FunctionTypeUtils.isFlux(this.outputType)) {
composedOutputType = ResolvableType.forClassWithGenerics(Flux.class, composedOutputType = ResolvableType.forClassWithGenerics(Flux.class, ResolvableType.forType(afterWrapper.outputType));
ResolvableType.forType(afterWrapper.outputType));
} }
else if (FunctionTypeUtils.isMono(this.outputType)) { else if (FunctionTypeUtils.isMono(this.outputType)) {
composedOutputType = ResolvableType.forClassWithGenerics(Mono.class, composedOutputType = ResolvableType.forClassWithGenerics(Mono.class, ResolvableType.forType(afterWrapper.outputType));
ResolvableType.forType(afterWrapper.outputType));
} }
else { else {
composedOutputType = ResolvableType.forType(afterWrapper.outputType); composedOutputType = ResolvableType.forType(afterWrapper.outputType);
} }
composedFunctionType = ResolvableType.forClassWithGenerics(Supplier.class, composedOutputType) composedFunctionType = ResolvableType.forClassWithGenerics(Supplier.class, composedOutputType).getType();
.getType();
} }
else if (this.outputType == null) { else if (this.outputType == null) {
throw new IllegalArgumentException("Can NOT compose anything with Consumer"); throw new IllegalArgumentException("Can NOT compose anything with Consumer");
} }
else { else {
composedFunctionType = ResolvableType composedFunctionType = ResolvableType.forClassWithGenerics(Function.class,
.forClassWithGenerics(Function.class, ResolvableType.forType(this.inputType), ResolvableType.forType(this.inputType),
ResolvableType.forType(((FunctionInvocationWrapper) after).outputType)) ResolvableType.forType(((FunctionInvocationWrapper) after).outputType)).getType();
.getType();
} }
String composedName = this.functionDefinition + "|" + afterWrapper.functionDefinition; String composedName = this.functionDefinition + "|" + afterWrapper.functionDefinition;
FunctionInvocationWrapper composedFunction = invocationWrapperInstance(composedName, rawComposedFunction, FunctionInvocationWrapper composedFunction = invocationWrapperInstance(composedName, rawComposedFunction, composedFunctionType);
composedFunctionType);
composedFunction.composed = true; composedFunction.composed = true;
return (Function<Object, V>) composedFunction; return (Function<Object, V>) composedFunction;
@@ -558,14 +548,12 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
*/ */
@Override @Override
public String toString() { public String toString() {
return this.functionDefinition return this.functionDefinition + (this.isComposed() ? "" : "<" + this.inputType + ", " + this.outputType + ">");
+ (this.isComposed() ? "" : "<" + this.inputType + ", " + this.outputType + ">");
} }
/** /**
* Returns true if this function wrapper represents a composed function. * Returns true if this function wrapper represents a composed function.
* @return true if this function wrapper represents a composed function otherwise * @return true if this function wrapper represents a composed function otherwise false
* false
*/ */
boolean isComposed() { boolean isComposed() {
return this.composed; return this.composed;
@@ -605,8 +593,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
} }
/** /**
* Will return Object.class if type is represented as TypeVariable(T) or * Will return Object.class if type is represented as TypeVariable(T) or WildcardType(?).
* WildcardType(?).
*/ */
private Type normalizeType(Type type) { private Type normalizeType(Type type) {
if (type != null) { if (type != null) {
@@ -619,13 +606,13 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
* *
*/ */
private Class<?> getRawClassFor(@Nullable Type type) { private Class<?> getRawClassFor(@Nullable Type type) {
return type instanceof TypeVariable || type instanceof WildcardType ? Object.class return type instanceof TypeVariable || type instanceof WildcardType
? Object.class
: FunctionTypeUtils.getRawType(type); : FunctionTypeUtils.getRawType(type);
} }
/** /**
* Will wrap the result in a Message if necessary and will copy input headers to * Will wrap the result in a Message if necessary and will copy input headers to the output message.
* the output message.
*/ */
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
private Object enrichInvocationResultIfNecessary(Object input, Object result) { private Object enrichInvocationResultIfNecessary(Object input, Object result) {
@@ -633,17 +620,14 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
if (result instanceof Message) { if (result instanceof Message) {
Map<String, Object> headersMap = (Map<String, Object>) ReflectionUtils Map<String, Object> headersMap = (Map<String, Object>) ReflectionUtils
.getField(SimpleFunctionRegistry.this.headersField, ((Message) result).getHeaders()); .getField(SimpleFunctionRegistry.this.headersField, ((Message) result).getHeaders());
this.sanitizeHeaders(((Message) input).getHeaders()) this.sanitizeHeaders(((Message) input).getHeaders()).forEach((k, v) -> headersMap.putIfAbsent(k, v));
.forEach((k, v) -> headersMap.putIfAbsent(k, v));
} }
else { else {
Message<Object> output = MessageBuilder.withPayload(result)
.copyHeaders(this.sanitizeHeaders(((Message) input).getHeaders())).build();
if (this.outputMessageHeaderEnricher != null) { if (this.outputMessageHeaderEnricher != null) {
result = this.outputMessageHeaderEnricher.enrich(output); result = this.outputMessageHeaderEnricher.apply((Message<?>) input, result);
} }
else { else {
result = output; result = MessageBuilder.withPayload(result).copyHeaders(this.sanitizeHeaders(((Message) input).getHeaders())).build();
} }
} }
} }
@@ -667,9 +651,9 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
* *
*/ */
private Object fluxifyInputIfNecessary(Object input) { private Object fluxifyInputIfNecessary(Object input) {
if (!(input instanceof Publisher) && this.isTypePublisher(this.inputType) if (!(input instanceof Publisher) && this.isTypePublisher(this.inputType) && !FunctionTypeUtils.isMultipleArgumentType(this.inputType)) {
&& !FunctionTypeUtils.isMultipleArgumentType(this.inputType)) { return input == null
return input == null ? FunctionTypeUtils.isMono(this.inputType) ? Mono.empty() : Flux.empty() ? FunctionTypeUtils.isMono(this.inputType) ? Mono.empty() : Flux.empty()
: FunctionTypeUtils.isMono(this.inputType) ? Mono.just(input) : Flux.just(input); : FunctionTypeUtils.isMono(this.inputType) ? Mono.just(input) : Flux.just(input);
} }
return input; return input;
@@ -683,24 +667,20 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
Object result; Object result;
if (!this.isTypePublisher(this.inputType) && convertedInput instanceof Publisher) { if (!this.isTypePublisher(this.inputType) && convertedInput instanceof Publisher) {
result = convertedInput instanceof Mono result = convertedInput instanceof Mono
? Mono.from((Publisher) convertedInput) ? Mono.from((Publisher) convertedInput).map(value -> this.invokeFunctionAndEnrichResultIfNecessary(value))
.map(value -> this.invokeFunctionAndEnrichResultIfNecessary(value)) .doOnError(ex -> logger.error("Failed to invoke function '" + this.functionDefinition + "'", (Throwable) ex))
.doOnError(ex -> logger.error( : Flux.from((Publisher) convertedInput).map(value -> this.invokeFunctionAndEnrichResultIfNecessary(value))
"Failed to invoke function '" + this.functionDefinition + "'", (Throwable) ex)) .doOnError(ex -> logger.error("Failed to invoke function '" + this.functionDefinition + "'", (Throwable) ex));
: Flux.from((Publisher) convertedInput)
.map(value -> this.invokeFunctionAndEnrichResultIfNecessary(value))
.doOnError(ex -> logger.error(
"Failed to invoke function '" + this.functionDefinition + "'", (Throwable) ex));
} }
else { else {
result = this.invokeFunctionAndEnrichResultIfNecessary(convertedInput); result = this.invokeFunctionAndEnrichResultIfNecessary(convertedInput);
if (result instanceof Flux) { if (result instanceof Flux) {
result = ((Flux) result).doOnError(ex -> logger result = ((Flux) result).doOnError(ex -> logger.error("Failed to invoke function '"
.error("Failed to invoke function '" + this.functionDefinition + "'", (Throwable) ex)); + this.functionDefinition + "'", (Throwable) ex));
} }
else if (result instanceof Mono) { else if (result instanceof Mono) {
result = ((Mono) result).doOnError(ex -> logger result = ((Mono) result).doOnError(ex -> logger.error("Failed to invoke function '"
.error("Failed to invoke function '" + this.functionDefinition + "'", (Throwable) ex)); + this.functionDefinition + "'", (Throwable) ex));
} }
} }
return result; return result;
@@ -727,8 +707,9 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
} }
Object result = ((Function) this.target).apply(inputValue); Object result = ((Function) this.target).apply(inputValue);
return value instanceof OriginalMessageHolder ? this.enrichInvocationResultIfNecessary( return value instanceof OriginalMessageHolder
((OriginalMessageHolder) value).getOriginalMessage(), result) : result; ? this.enrichInvocationResultIfNecessary(((OriginalMessageHolder) value).getOriginalMessage(), result)
: result;
} }
/* /*
@@ -739,20 +720,20 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
Object result = null; Object result = null;
if (this.isTypePublisher(this.inputType)) { if (this.isTypePublisher(this.inputType)) {
if (convertedInput instanceof Flux) { if (convertedInput instanceof Flux) {
result = ((Flux) convertedInput).transform(flux -> { result = ((Flux) convertedInput)
flux = Flux.from((Publisher) flux) .transform(flux -> {
.map(v -> this.extractValueFromOriginalValueHolderIfNecessary(v)); flux = Flux.from((Publisher) flux).map(v -> this.extractValueFromOriginalValueHolderIfNecessary(v));
((Consumer) this.target).accept(flux); ((Consumer) this.target).accept(flux);
return Mono.ignoreElements((Flux) flux); return Mono.ignoreElements((Flux) flux);
}).then(); }).then();
} }
else { else {
result = ((Mono) convertedInput).transform(mono -> { result = ((Mono) convertedInput)
mono = Mono.from((Publisher) mono) .transform(mono -> {
.map(v -> this.extractValueFromOriginalValueHolderIfNecessary(v)); mono = Mono.from((Publisher) mono).map(v -> this.extractValueFromOriginalValueHolderIfNecessary(v));
((Consumer) this.target).accept(mono); ((Consumer) this.target).accept(mono);
return Mono.ignoreElements((Flux) mono); return Mono.ignoreElements((Flux) mono);
}).then(); }).then();
} }
} }
else if (convertedInput instanceof Publisher) { else if (convertedInput instanceof Publisher) {
@@ -790,14 +771,12 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
} }
return parsedArgumentValues; return parsedArgumentValues;
} }
throw new UnsupportedOperationException( throw new UnsupportedOperationException("At the moment only Tuple-based function are supporting multiple arguments");
"At the moment only Tuple-based function are supporting multiple arguments");
} }
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
private boolean isInputConversionNecessary(Object input, Type type) { private boolean isInputConversionNecessary(Object input, Type type) {
if (type == null || this.getRawClassFor(type) == Void.class || this.target instanceof RoutingFunction if (type == null || this.getRawClassFor(type) == Void.class || this.target instanceof RoutingFunction || this.isComposed()) {
|| this.isComposed()) {
if (this.getRawClassFor(type) == Void.class) { if (this.getRawClassFor(type) == Void.class) {
if (input instanceof Message) { if (input instanceof Message) {
input = ((Message) input).getPayload(); input = ((Message) input).getPayload();
@@ -811,7 +790,6 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
} }
return true; return true;
} }
/* /*
* *
*/ */
@@ -835,13 +813,13 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
convertedInput = Tuples.fromArray(convertedInputs); convertedInput = Tuples.fromArray(convertedInputs);
} }
else if (this.skipInputConversion) { else if (this.skipInputConversion) {
convertedInput = this.isInputTypeMessage() ? input convertedInput = this.isInputTypeMessage()
? input
: new OriginalMessageHolder(((Message) input).getPayload(), (Message<?>) input); : new OriginalMessageHolder(((Message) input).getPayload(), (Message<?>) input);
} }
else if (input instanceof Message) { else if (input instanceof Message) {
if (((Message) input).getPayload().getClass().getName() if (((Message) input).getPayload().getClass().getName().equals("org.springframework.kafka.support.KafkaNull")
.equals("org.springframework.kafka.support.KafkaNull") && !this.isInputTypeMessage()) { // TODO && !this.isInputTypeMessage()) { //TODO rework
// rework
return null; return null;
} }
@@ -849,12 +827,12 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
convertedInput = this.convertInputMessageIfNecessary((Message) input, type); convertedInput = this.convertInputMessageIfNecessary((Message) input, type);
if (convertedInput == null) { // give ConversionService a chance if (convertedInput == null) { // give ConversionService a chance
convertedInput = this.convertNonMessageInputIfNecessary(type, ((Message) input).getPayload(), convertedInput = this.convertNonMessageInputIfNecessary(type, ((Message) input).getPayload(), false);
false);
} }
if (convertedInput != null && !FunctionTypeUtils.isMultipleArgumentType(this.inputType)) { if (convertedInput != null && !FunctionTypeUtils.isMultipleArgumentType(this.inputType)) {
convertedInput = !convertedInput.equals(input) convertedInput = !convertedInput.equals(input)
? new OriginalMessageHolder(convertedInput, (Message<?>) input) : convertedInput; ? new OriginalMessageHolder(convertedInput, (Message<?>) input)
: convertedInput;
} }
if (convertedInput != null && logger.isDebugEnabled()) { if (convertedInput != null && logger.isDebugEnabled()) {
logger.debug("Converted Message: " + input + " to: " + convertedInput); logger.debug("Converted Message: " + input + " to: " + convertedInput);
@@ -875,17 +853,16 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
} }
/** /**
* This is an optional conversion which would only happen if * This is an optional conversion which would only happen if `expected-content-type` is
* `expected-content-type` is set as a header in a message or explicitly provided * set as a header in a message or explicitly provided as part of the lookup.
* as part of the lookup.
*/ */
private Object convertOutputIfNecessary(Object output, Type type, String[] contentType) { private Object convertOutputIfNecessary(Object output, Type type, String[] contentType) {
if (this.skipOutputConversion) { if (this.skipOutputConversion) {
return output; return output;
} }
if (output instanceof Message && !this.containsRetainMessageSignalInHeaders((Message) output)) { if (output instanceof Message && !this.containsRetainMessageSignalInHeaders((Message) output)) {
if (!FunctionTypeUtils.isMessage(type) || (FunctionTypeUtils.isMessage(type) if (!FunctionTypeUtils.isMessage(type) ||
&& Collection.class.isAssignableFrom(FunctionTypeUtils.getRawType(type)))) { (FunctionTypeUtils.isMessage(type) && Collection.class.isAssignableFrom(FunctionTypeUtils.getRawType(type)))) {
output = ((Message) output).getPayload(); output = ((Message) output).getPayload();
} }
} }
@@ -905,16 +882,13 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
convertedOutput = this.convertOutputPublisherIfNecessary((Publisher) output, type, contentType); convertedOutput = this.convertOutputPublisherIfNecessary((Publisher) output, type, contentType);
} }
else if (output instanceof Message) { else if (output instanceof Message) {
convertedOutput = this.convertOutputMessageIfNecessary(output, convertedOutput = this.convertOutputMessageIfNecessary(output, ObjectUtils.isEmpty(contentType) ? null : contentType[0]);
ObjectUtils.isEmpty(contentType) ? null : contentType[0]);
} }
else if (output instanceof Collection && this.isOutputTypeMessage()) { else if (output instanceof Collection && this.isOutputTypeMessage()) {
convertedOutput = this.convertMultipleOutputValuesIfNecessary(output, convertedOutput = this.convertMultipleOutputValuesIfNecessary(output, ObjectUtils.isEmpty(contentType) ? null : contentType);
ObjectUtils.isEmpty(contentType) ? null : contentType);
} }
else if (ObjectUtils.isArray(output) && !(output instanceof byte[])) { else if (ObjectUtils.isArray(output) && !(output instanceof byte[])) {
convertedOutput = this.convertMultipleOutputValuesIfNecessary(output, convertedOutput = this.convertMultipleOutputValuesIfNecessary(output, ObjectUtils.isEmpty(contentType) ? null : contentType);
ObjectUtils.isEmpty(contentType) ? null : contentType);
} }
else { else {
convertedOutput = messageConverter.toMessage(output, convertedOutput = messageConverter.toMessage(output,
@@ -925,16 +899,15 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
} }
/** /**
* Will check if message contains any of the headers that are considered to serve * Will check if message contains any of the headers that are considered to serve as
* as signals to retain output as Message (regardless of the output type of * signals to retain output as Message (regardless of the output type of function).
* function). At this moment presence of 'scf-func-name' header or any header that * At this moment presence of 'scf-func-name' header or any header that begins with `lambda'
* begins with `lambda' (use by AWS) will result in this method returning true. * (use by AWS) will result in this method returning true.
*/ */
/* /*
* TODO we need to investigate if this could be extracted into some type of * TODO we need to investigate if this could be extracted into some type of strategy since at
* strategy since at the pure core level there is no case for this to ever be * the pure core level there is no case for this to ever be true. In fact today it is only AWS Lambda
* true. In fact today it is only AWS Lambda case that requires it since it may * case that requires it since it may contain forwarding url
* contain forwarding url
*/ */
private boolean containsRetainMessageSignalInHeaders(Message message) { private boolean containsRetainMessageSignalInHeaders(Message message) {
if (new CloudEventAttributes(message.getHeaders()).isValidCloudEvent()) { if (new CloudEventAttributes(message.getHeaders()).isValidCloudEvent()) {
@@ -942,7 +915,8 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
} }
else { else {
for (String headerName : message.getHeaders().keySet()) { for (String headerName : message.getHeaders().keySet()) {
if (headerName.startsWith("lambda") || headerName.startsWith("scf-func-name")) { if (headerName.startsWith("lambda") ||
headerName.startsWith("scf-func-name")) {
return true; return true;
} }
} }
@@ -967,7 +941,8 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
convertedInput = SimpleFunctionRegistry.this.jsonMapper.fromJson(input, inputType); convertedInput = SimpleFunctionRegistry.this.jsonMapper.fromJson(input, inputType);
} }
} }
else if (SimpleFunctionRegistry.this.conversionService != null && !rawInputType.equals(input.getClass()) else if (SimpleFunctionRegistry.this.conversionService != null
&& !rawInputType.equals(input.getClass())
&& SimpleFunctionRegistry.this.conversionService.canConvert(input.getClass(), rawInputType)) { && SimpleFunctionRegistry.this.conversionService.canConvert(input.getClass(), rawInputType)) {
convertedInput = SimpleFunctionRegistry.this.conversionService.convert(input, rawInputType); convertedInput = SimpleFunctionRegistry.this.conversionService.convert(input, rawInputType);
} }
@@ -981,8 +956,10 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
* *
*/ */
private boolean isWrapConvertedInputInMessage(Object convertedInput) { private boolean isWrapConvertedInputInMessage(Object convertedInput) {
return this.inputType != null && FunctionTypeUtils.isMessage(this.inputType) return this.inputType != null
&& !(convertedInput instanceof Message) && !(convertedInput instanceof Publisher) && FunctionTypeUtils.isMessage(this.inputType)
&& !(convertedInput instanceof Message)
&& !(convertedInput instanceof Publisher)
&& !(convertedInput instanceof OriginalMessageHolder); && !(convertedInput instanceof OriginalMessageHolder);
} }
@@ -990,8 +967,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
* *
*/ */
private Type extractActualValueTypeIfNecessary(Type type) { private Type extractActualValueTypeIfNecessary(Type type) {
if (type instanceof ParameterizedType if (type instanceof ParameterizedType && (FunctionTypeUtils.isPublisher(type) || FunctionTypeUtils.isMessage(type))) {
&& (FunctionTypeUtils.isPublisher(type) || FunctionTypeUtils.isMessage(type))) {
return FunctionTypeUtils.getGenericType(type); return FunctionTypeUtils.getGenericType(type);
} }
return type; return type;
@@ -1032,12 +1008,10 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
if (this.isInputTypeMessage()) { if (this.isInputTypeMessage()) {
if (convertedInput == null) { if (convertedInput == null) {
/* /*
* In the event conversion was unsuccessful we simply return the * In the event conversion was unsuccessful we simply return the original un-converted message.
* original un-converted message. This will help to deal with issues * This will help to deal with issues like KafkaNull and others. However if this was not the intention
* like KafkaNull and others. However if this was not the intention of * of the developer, this would be discovered early in the development process where the
* the developer, this would be discovered early in the development * additional message converter could be added to facilitate the conversion.
* process where the additional message converter could be added to
* facilitate the conversion.
*/ */
logger.info("Input type conversion of payload " + message.getPayload() + " resulted in 'null'. " logger.info("Input type conversion of payload " + message.getPayload() + " resulted in 'null'. "
+ "Will use the original message as input."); + "Will use the original message as input.");
@@ -1045,8 +1019,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
} }
else { else {
if (!(convertedInput instanceof Message)) { if (!(convertedInput instanceof Message)) {
convertedInput = MessageBuilder.withPayload(convertedInput).copyHeaders(message.getHeaders()) convertedInput = MessageBuilder.withPayload(convertedInput).copyHeaders(message.getHeaders()).build();
.build();
} }
} }
} }
@@ -1061,10 +1034,10 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
Object[] multipleValueArguments = this.parseMultipleValueArguments(output, outputTypes.length); Object[] multipleValueArguments = this.parseMultipleValueArguments(output, outputTypes.length);
Object[] convertedOutputs = new Object[outputTypes.length]; Object[] convertedOutputs = new Object[outputTypes.length];
for (int i = 0; i < multipleValueArguments.length; i++) { for (int i = 0; i < multipleValueArguments.length; i++) {
String[] ctToUse = !ObjectUtils.isEmpty(contentType) ? new String[] { contentType[i] } String[] ctToUse = !ObjectUtils.isEmpty(contentType)
: new String[] { "application/json" }; ? new String[]{contentType[i]}
Object convertedInput = this.convertOutputIfNecessary(multipleValueArguments[i], outputTypes[i], : new String[] {"application/json"};
ctToUse); Object convertedInput = this.convertOutputIfNecessary(multipleValueArguments[i], outputTypes[i], ctToUse);
convertedOutputs[i] = convertedInput; convertedOutputs[i] = convertedInput;
} }
return Tuples.fromArray(convertedOutputs); return Tuples.fromArray(convertedOutputs);
@@ -1077,18 +1050,15 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
private Object convertOutputMessageIfNecessary(Object output, String expectedOutputContetntType) { private Object convertOutputMessageIfNecessary(Object output, String expectedOutputContetntType) {
Map<String, Object> headersMap = (Map<String, Object>) ReflectionUtils Map<String, Object> headersMap = (Map<String, Object>) ReflectionUtils
.getField(SimpleFunctionRegistry.this.headersField, ((Message) output).getHeaders()); .getField(SimpleFunctionRegistry.this.headersField, ((Message) output).getHeaders());
String contentType = ((Message) output).getHeaders() String contentType = ((Message) output).getHeaders().containsKey(FunctionProperties.EXPECT_CONTENT_TYPE_HEADER)
.containsKey(FunctionProperties.EXPECT_CONTENT_TYPE_HEADER) ? (String) ((Message) output).getHeaders().get(FunctionProperties.EXPECT_CONTENT_TYPE_HEADER)
? (String) ((Message) output).getHeaders()
.get(FunctionProperties.EXPECT_CONTENT_TYPE_HEADER)
: expectedOutputContetntType; : expectedOutputContetntType;
if (StringUtils.hasText(contentType)) { if (StringUtils.hasText(contentType)) {
String[] expectedContentTypes = StringUtils.delimitedListToStringArray(contentType, ","); String[] expectedContentTypes = StringUtils.delimitedListToStringArray(contentType, ",");
for (String expectedContentType : expectedContentTypes) { for (String expectedContentType : expectedContentTypes) {
headersMap.put(MessageHeaders.CONTENT_TYPE, expectedContentType); headersMap.put(MessageHeaders.CONTENT_TYPE, expectedContentType);
Object result = messageConverter.toMessage(((Message) output).getPayload(), Object result = messageConverter.toMessage(((Message) output).getPayload(), ((Message) output).getHeaders());
((Message) output).getHeaders());
if (result != null) { if (result != null) {
return result; return result;
} }
@@ -1102,12 +1072,9 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
*/ */
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
private Object convertMultipleOutputValuesIfNecessary(Object output, String[] contentType) { private Object convertMultipleOutputValuesIfNecessary(Object output, String[] contentType) {
Collection outputCollection = ObjectUtils.isArray(output) ? CollectionUtils.arrayToList(output) Collection outputCollection = ObjectUtils.isArray(output) ? CollectionUtils.arrayToList(output) : (Collection) output;
: (Collection) output; Collection convertedOutputCollection = outputCollection instanceof List ? new ArrayList<>() : new TreeSet<>();
Collection convertedOutputCollection = outputCollection instanceof List ? new ArrayList<>() Type type = this.isOutputTypeMessage() ? FunctionTypeUtils.getGenericType(this.outputType) : this.outputType;
: new TreeSet<>();
Type type = this.isOutputTypeMessage() ? FunctionTypeUtils.getGenericType(this.outputType)
: this.outputType;
for (Object outToConvert : outputCollection) { for (Object outToConvert : outputCollection) {
Object result = this.convertOutputIfNecessary(outToConvert, type, contentType); Object result = this.convertOutputIfNecessary(outToConvert, type, contentType);
Assert.notNull(result, () -> "Failed to convert output '" + outToConvert + "'"); Assert.notNull(result, () -> "Failed to convert output '" + outToConvert + "'");
@@ -1139,22 +1106,19 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
* *
*/ */
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
private Object convertOutputPublisherIfNecessary(Publisher publisher, Type type, private Object convertOutputPublisherIfNecessary(Publisher publisher, Type type, String[] expectedOutputContentType) {
String[] expectedOutputContentType) {
return publisher instanceof Mono return publisher instanceof Mono
? Mono.from(publisher).map(v -> this.convertOutputIfNecessary(v, type, expectedOutputContentType)) ? Mono.from(publisher).map(v -> this.convertOutputIfNecessary(v, type, expectedOutputContentType))
.doOnError(ex -> logger.error("Failed to convert output", (Throwable) ex)) .doOnError(ex -> logger.error("Failed to convert output", (Throwable) ex))
: Flux.from(publisher).map(v -> this.convertOutputIfNecessary(v, type, expectedOutputContentType)) : Flux.from(publisher).map(v -> this.convertOutputIfNecessary(v, type, expectedOutputContentType))
.doOnError(ex -> logger.error("Failed to convert output", (Throwable) ex)); .doOnError(ex -> logger.error("Failed to convert output", (Throwable) ex));
} }
} }
/** /**
* *
*/ */
private static final class OriginalMessageHolder { private static final class OriginalMessageHolder {
private final Object value; private final Object value;
private final Message<?> originalMessage; private final Message<?> originalMessage;
@@ -1171,7 +1135,5 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
public Message<?> getOriginalMessage() { public Message<?> getOriginalMessage() {
return this.originalMessage; return this.originalMessage;
} }
} }
} }

34
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/CloudEventAutoConfiguration.java
View File

@@ -1,34 +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.config;
import org.springframework.cloud.function.cloudevent.CloudEventOutputMessageHeaderEnricher;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @author Dave Syer
*
*/
@Configuration(proxyBeanMethods = false)
public class CloudEventAutoConfiguration {
@Bean
public CloudEventOutputMessageHeaderEnricher cloudEventOutputMessageHeaderEnricher() {
return new CloudEventOutputMessageHeaderEnricher();
}
}

44
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/message/CompositeOutputMessageHeaderEnricher.java
View File

@@ -1,44 +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.message;
import java.util.List;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
/**
* @author Dave Syer
*
*/
public class CompositeOutputMessageHeaderEnricher implements OutputMessageHeaderEnricher {
private final List<OutputMessageHeaderEnricher> delegates;
public CompositeOutputMessageHeaderEnricher(List<OutputMessageHeaderEnricher> delegates) {
this.delegates = delegates;
}
@Override
public Message<?> enrich(Message<?> output) {
Message<?> result = MessageBuilder.withPayload(output.getPayload()).copyHeaders(output.getHeaders()).build();
for (OutputMessageHeaderEnricher enricher : delegates) {
result = enricher.enrich(result);
}
return result;
}
}

28
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/message/OutputMessageHeaderEnricher.java
View File

@@ -1,28 +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.message;
import org.springframework.messaging.Message;
/**
* @author Dave Syer
*
*/
public interface OutputMessageHeaderEnricher {
Message<?> enrich(Message<?> output);
}

3
spring-cloud-function-context/src/main/resources/META-INF/spring.factories
View File

@@ -1,6 +1,5 @@
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.function.context.config.ContextFunctionCatalogAutoConfiguration,\ org.springframework.cloud.function.context.config.ContextFunctionCatalogAutoConfiguration
org.springframework.cloud.function.context.config.CloudEventAutoConfiguration
org.springframework.cloud.function.context.WrapperDetector=\ org.springframework.cloud.function.context.WrapperDetector=\
org.springframework.cloud.function.context.config.FluxWrapperDetector org.springframework.cloud.function.context.config.FluxWrapperDetector
org.springframework.context.ApplicationContextInitializer=\ org.springframework.context.ApplicationContextInitializer=\