From 280b8844c765b3490647503eb2e3b2f483bb85b5 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Wed, 12 May 2021 08:27:26 +0200 Subject: [PATCH] GH-676 Add initial support for mapping/re-mapping input headers Add BeanResolver --- .../function/context/FunctionProperties.java | 70 +++++++++- .../BeanFactoryAwareFunctionRegistry.java | 4 +- .../context/catalog/InputEnricher.java | 74 ++++++++++ .../catalog/SimpleFunctionRegistry.java | 35 ++++- ...ntextFunctionCatalogAutoConfiguration.java | 4 +- .../context/FunctionPropertiesTests.java | 132 ++++++++++++++++++ 6 files changed, 311 insertions(+), 8 deletions(-) create mode 100644 spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/InputEnricher.java create mode 100644 spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/FunctionPropertiesTests.java diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionProperties.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionProperties.java index 0108d7f4f..2bda325e6 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionProperties.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2019 the original author or authors. + * Copyright 2019-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. @@ -16,7 +16,16 @@ package org.springframework.cloud.function.context; +import java.util.Collections; +import java.util.Map; +import java.util.Map.Entry; + +import org.springframework.beans.BeansException; import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.context.EnvironmentAware; +import org.springframework.core.env.Environment; /** * @@ -25,7 +34,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties; * */ @ConfigurationProperties(prefix = FunctionProperties.PREFIX) -public class FunctionProperties { +public class FunctionProperties implements EnvironmentAware, ApplicationContextAware { /** * The name prefix for properties defined by this properties class. @@ -55,15 +64,53 @@ public class FunctionProperties { */ private String definition; + private Map configuration; private String expectedContentType; + private Environment environment; + + private ApplicationContext applicationContext; + + public Map getConfiguration() { + return configuration; + } + /** * SpEL expression which should result in function definition (e.g., function name or composition instruction). * NOTE: SpEL evaluation context's root object is the input argument (e.g., Message). */ private String routingExpression; + public void setConfiguration(Map configuration) { + for (Entry entry : configuration.entrySet()) { + String propertyX = "spring.cloud.function.configuration." + entry.getKey() + ".input-header-mapping-expression."; + String propertyY = "spring.cloud.function.configuration." + entry.getKey() + ".inputHeaderMappingExpression."; + Map headerMapping = entry.getValue().getInputHeaderMappingExpression(); + for (Object k : headerMapping.keySet()) { + if (this.environment.containsProperty(propertyX + k) || this.environment.containsProperty(propertyY + k)) { + Map originalMapping = entry.getValue().getInputHeaderMappingExpression(); + entry.getValue().setInputHeaderMappingExpression(Collections.singletonMap("0", originalMapping)); + break; + } + else { + break; + } + } + } + this.configuration = configuration; + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) + throws BeansException { + this.applicationContext = applicationContext; + } + + public ApplicationContext getApplicationContext() { + return this.applicationContext; + } + public String getDefinition() { return definition; } @@ -87,4 +134,23 @@ public class FunctionProperties { public void setExpectedContentType(String expectedContentType) { this.expectedContentType = expectedContentType; } + + @Override + public void setEnvironment(Environment environment) { + this.environment = environment; + } + + public static class FunctionConfigurationProperties { + + private Map inputHeaderMappingExpression; + + public Map getInputHeaderMappingExpression() { + return inputHeaderMappingExpression; + } + + public void setInputHeaderMappingExpression(Map inputHeaderMappingExpression) { + this.inputHeaderMappingExpression = inputHeaderMappingExpression; + } + + } } diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistry.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistry.java index c893c96bc..942ae7387 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistry.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistry.java @@ -55,8 +55,8 @@ public class BeanFactoryAwareFunctionRegistry extends SimpleFunctionRegistry imp private GenericApplicationContext applicationContext; public BeanFactoryAwareFunctionRegistry(ConversionService conversionService, CompositeMessageConverter messageConverter, - JsonMapper jsonMapper, @Nullable FunctionInvocationHelper> functionInvocationHelper) { - super(conversionService, messageConverter, jsonMapper, functionInvocationHelper); + JsonMapper jsonMapper, @Nullable FunctionProperties functionProperties, @Nullable FunctionInvocationHelper> functionInvocationHelper) { + super(conversionService, messageConverter, jsonMapper, functionProperties, functionInvocationHelper); } @Override diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/InputEnricher.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/InputEnricher.java new file mode 100644 index 000000000..2cc95f411 --- /dev/null +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/InputEnricher.java @@ -0,0 +1,74 @@ +/* + * Copyright 2021-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. + * 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.Map; +import java.util.Map.Entry; +import java.util.function.Function; + +import org.springframework.context.expression.MapAccessor; +import org.springframework.expression.BeanResolver; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.expression.spel.support.StandardEvaluationContext; +import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.Assert; + +/** + * + * @author Oleg Zhurakousky + * + * @since 3.1.3 + * + */ +public class InputEnricher implements Function { + + private final Map> headerExpressions; + + private final SpelExpressionParser spelParser = new SpelExpressionParser(); + + private final StandardEvaluationContext evalContext = new StandardEvaluationContext(); + + @SuppressWarnings({ "rawtypes", "unchecked" }) + public InputEnricher(Map headerExpressions, @Nullable BeanResolver beanResolver) { + Assert.notEmpty(headerExpressions, "'headerExpressions' must not be null or empty"); + this.headerExpressions = headerExpressions; + this.evalContext.addPropertyAccessor(new MapAccessor()); + if (beanResolver != null) { + this.evalContext.setBeanResolver(beanResolver); + } + } + + @Override + public Object apply(Object input) { + if (input instanceof Message) { + MessageBuilder messageBuilder = MessageBuilder.fromMessage((Message) input); + Map mappings = this.headerExpressions.get("0"); + for (Entry keyValueExpressionEntry : mappings.entrySet()) { + Expression expression = this.spelParser.parseExpression(keyValueExpressionEntry.getValue()); + Object value = expression.getValue(this.evalContext, input, Object.class); + messageBuilder.setHeader(keyValueExpressionEntry.getKey(), value); + } + input = messageBuilder.build(); + } + + return input; + } + +} 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 5c4e1e8b8..2931ac7d6 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 @@ -49,11 +49,13 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.cloud.function.cloudevent.CloudEventMessageUtils; import org.springframework.cloud.function.context.FunctionCatalog; import org.springframework.cloud.function.context.FunctionProperties; +import org.springframework.cloud.function.context.FunctionProperties.FunctionConfigurationProperties; import org.springframework.cloud.function.context.FunctionRegistration; import org.springframework.cloud.function.context.FunctionRegistry; import org.springframework.cloud.function.context.config.RoutingFunction; import org.springframework.cloud.function.core.FunctionInvocationHelper; import org.springframework.cloud.function.json.JsonMapper; +import org.springframework.context.expression.BeanFactoryResolver; import org.springframework.core.ResolvableType; import org.springframework.core.convert.ConversionService; import org.springframework.expression.Expression; @@ -100,10 +102,13 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect private final FunctionInvocationHelper> functionInvocationHelper; + private final FunctionProperties functionProperties; + @Autowired(required = false) private FunctionAroundWrapper functionAroundWrapper; public SimpleFunctionRegistry(ConversionService conversionService, CompositeMessageConverter messageConverter, JsonMapper jsonMapper, + @Nullable FunctionProperties functionProperties, @Nullable FunctionInvocationHelper> functionInvocationHelper) { Assert.notNull(messageConverter, "'messageConverter' must not be null"); Assert.notNull(jsonMapper, "'jsonMapper' must not be null"); @@ -113,6 +118,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect this.headersField = ReflectionUtils.findField(MessageHeaders.class, "headers"); this.headersField.setAccessible(true); this.functionInvocationHelper = functionInvocationHelper; + this.functionProperties = functionProperties; } @Override @@ -123,7 +129,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect } public SimpleFunctionRegistry(ConversionService conversionService, CompositeMessageConverter messageConverter, JsonMapper jsonMapper) { - this(conversionService, messageConverter, jsonMapper, null); + this(conversionService, messageConverter, jsonMapper, null, null); } @SuppressWarnings("unchecked") @@ -277,6 +283,8 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect FunctionInvocationWrapper andThenFunction = invocationWrapperInstance(functionName, function.getTarget(), function.inputType, function.outputType); composedFunction = (FunctionInvocationWrapper) composedFunction.andThen((Function) andThenFunction); + composedFunction = this.enrichInputIfNecessary(composedFunction); + } this.wrappedFunctionDefinitions.put(composedFunction.functionDefinition, composedFunction); } @@ -287,6 +295,29 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect return composedFunction; } + private FunctionInvocationWrapper enrichInputIfNecessary(FunctionInvocationWrapper composedFunction) { + if (this.functionProperties == null) { + return composedFunction; + } + String functionDefinition = composedFunction.getFunctionDefinition(); + Map configurationProperties = this.functionProperties.getConfiguration(); + if (!CollectionUtils.isEmpty(configurationProperties)) { + FunctionConfigurationProperties configuration = configurationProperties + .get(functionDefinition.replace("|", "").replace(",", "")); + if (!CollectionUtils.isEmpty(configuration.getInputHeaderMappingExpression())) { + BeanFactoryResolver beanResolver = this.functionProperties.getApplicationContext() != null + ? new BeanFactoryResolver(this.functionProperties.getApplicationContext()) + : null; + InputEnricher enricher = new InputEnricher(configuration.getInputHeaderMappingExpression(), beanResolver); + FunctionInvocationWrapper w = new FunctionInvocationWrapper("headerEnricher", enricher, Message.class, Message.class); + composedFunction = (FunctionInvocationWrapper) w.andThen((Function) composedFunction); + composedFunction.functionDefinition = functionDefinition; + } + } + + return composedFunction; + } + /* * */ @@ -315,7 +346,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect private final Type outputType; - private final String functionDefinition; + private String functionDefinition; private boolean composed; diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java index 20527af40..ca2d437b8 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java @@ -125,8 +125,8 @@ public class ContextFunctionCatalogAutoConfiguration { ((CloudEventsFunctionInvocationHelper) functionInvocationHelper).setMessageConverter(messageConverter); } } - - return new BeanFactoryAwareFunctionRegistry(conversionService, messageConverter, jsonMapper, functionInvocationHelper); + FunctionProperties functionProperties = context.getBean(FunctionProperties.class); + return new BeanFactoryAwareFunctionRegistry(conversionService, messageConverter, jsonMapper, functionProperties, functionInvocationHelper); } @Bean(RoutingFunction.FUNCTION_NAME) diff --git a/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/FunctionPropertiesTests.java b/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/FunctionPropertiesTests.java new file mode 100644 index 000000000..64a8043f1 --- /dev/null +++ b/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/FunctionPropertiesTests.java @@ -0,0 +1,132 @@ +/* + * Copyright 2019-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. + * 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.Map; +import java.util.function.Function; + +import org.junit.jupiter.api.Test; + +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.function.context.FunctionProperties.FunctionConfigurationProperties; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import static org.assertj.core.api.Assertions.assertThat; + +public class FunctionPropertiesTests { + + @Test + public void testInputHeaderMappingPropertyWithIndex() throws Exception { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run( + "--logging.level.org.springframework.cloud.function=DEBUG", + "--spring.main.lazy-initialization=true", + "--spring.cloud.function.definition=echo", + "--spring.cloud.function.configuration.echo.input-header-mapping-expression[0].key1=hello1", + "--spring.cloud.function.configuration.echo.input-header-mapping-expression[0].key2=hello2", + "--spring.cloud.function.configuration.echo.input-header-mapping-expression[1].key12=hello12")) { + FunctionProperties functionProperties = context + .getBean(FunctionProperties.class); + FunctionConfigurationProperties configuration = functionProperties + .getConfiguration().get("echo"); + assertThat(configuration.getInputHeaderMappingExpression()).containsKey("0"); + assertThat(configuration.getInputHeaderMappingExpression()).containsKey("1"); + } + } + + @Test + public void testInputHeaderMappingPropertyWithoutIndex() throws Exception { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run( + "--logging.level.org.springframework.cloud.function=DEBUG", + "--spring.main.lazy-initialization=true", + "--spring.cloud.function.definition=echo", + "--spring.cloud.function.configuration.echo.input-header-mapping-expression.key1=hello1", + "--spring.cloud.function.configuration.echo.input-header-mapping-expression.key2=hello2")) { + FunctionProperties functionProperties = context + .getBean(FunctionProperties.class); + FunctionConfigurationProperties configuration = functionProperties + .getConfiguration().get("echo"); + assertThat(configuration.getInputHeaderMappingExpression()).containsKey("0"); + } + } + + @Test + public void testInputHeaderMappingPropertyWithCompositionWithIndex() throws Exception { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run( + "--logging.level.org.springframework.cloud.function=DEBUG", + "--spring.main.lazy-initialization=true", +// "--spring.cloud.function.definition=echo|foo", + "--spring.cloud.function.configuration.echofoo.input-header-mapping-expression[0].key1=hello1", + "--spring.cloud.function.configuration.echofoo.input-header-mapping-expression[0].key2=hello2", + "--spring.cloud.function.configuration.echofoo.input-header-mapping-expression[1].key12=hello12")) { + FunctionProperties functionProperties = context + .getBean(FunctionProperties.class); + FunctionConfigurationProperties configuration = functionProperties + .getConfiguration().get("echofoo"); + Map keyValueExpression = (Map) configuration.getInputHeaderMappingExpression().get("0"); + + +// FunctionCatalog functionCatalog = context.getBean(FunctionCatalog.class); +// FunctionInvocationWrapper function = functionCatalog.lookup("echo|foo"); +// System.out.println(function.apply(new GenericMessage("helo"))); + + + +// System.out.println(keyValueExpression.get("key1")); +// assertThat(configuration.getInputHeaderMappingExpression()).containsKey("0"); +// assertThat(configuration.getInputHeaderMappingExpression()).containsKey("1"); + } + } + + @Test + public void testInputHeaderMappingPropertyWithCompositionWithoutIndex() throws Exception { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run( + "--logging.level.org.springframework.cloud.function=DEBUG", + "--spring.main.lazy-initialization=true", + "--spring.cloud.function.definition=echo|foo", + "--spring.cloud.function.configuration.echofoo.input-header-mapping-expression.key1=hello1", + "--spring.cloud.function.configuration.echofoo.input-header-mapping-expression.key2=hello2")) { + FunctionProperties functionProperties = context + .getBean(FunctionProperties.class); + FunctionConfigurationProperties configuration = functionProperties + .getConfiguration().get("echofoo"); + assertThat(configuration.getInputHeaderMappingExpression()).containsKey("0"); + } + } + + @EnableAutoConfiguration + @Configuration + protected static class SampleFunctionConfiguration { + + @Bean + public Function echo() { + return x -> x; + } + + @Bean + public Function foo() { + return x -> x; + } + } +}