GH-676 Improvements to header enrichment
This commit is contained in:
@@ -283,9 +283,8 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
|
|||||||
FunctionInvocationWrapper andThenFunction =
|
FunctionInvocationWrapper andThenFunction =
|
||||||
invocationWrapperInstance(functionName, function.getTarget(), function.inputType, function.outputType);
|
invocationWrapperInstance(functionName, function.getTarget(), function.inputType, function.outputType);
|
||||||
composedFunction = (FunctionInvocationWrapper) composedFunction.andThen((Function<Object, Object>) andThenFunction);
|
composedFunction = (FunctionInvocationWrapper) composedFunction.andThen((Function<Object, Object>) andThenFunction);
|
||||||
composedFunction = this.enrichInputIfNecessary(composedFunction);
|
|
||||||
|
|
||||||
}
|
}
|
||||||
|
composedFunction = this.enrichInputIfNecessary(composedFunction);
|
||||||
this.wrappedFunctionDefinitions.put(composedFunction.functionDefinition, composedFunction);
|
this.wrappedFunctionDefinitions.put(composedFunction.functionDefinition, composedFunction);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -304,17 +303,18 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
|
|||||||
if (!CollectionUtils.isEmpty(configurationProperties)) {
|
if (!CollectionUtils.isEmpty(configurationProperties)) {
|
||||||
FunctionConfigurationProperties configuration = configurationProperties
|
FunctionConfigurationProperties configuration = configurationProperties
|
||||||
.get(functionDefinition.replace("|", "").replace(",", ""));
|
.get(functionDefinition.replace("|", "").replace(",", ""));
|
||||||
if (!CollectionUtils.isEmpty(configuration.getInputHeaderMappingExpression())) {
|
if (configuration != null) {
|
||||||
BeanFactoryResolver beanResolver = this.functionProperties.getApplicationContext() != null
|
if (!CollectionUtils.isEmpty(configuration.getInputHeaderMappingExpression())) {
|
||||||
? new BeanFactoryResolver(this.functionProperties.getApplicationContext())
|
BeanFactoryResolver beanResolver = this.functionProperties.getApplicationContext() != null
|
||||||
: null;
|
? new BeanFactoryResolver(this.functionProperties.getApplicationContext())
|
||||||
InputEnricher enricher = new InputEnricher(configuration.getInputHeaderMappingExpression(), beanResolver);
|
: null;
|
||||||
FunctionInvocationWrapper w = new FunctionInvocationWrapper("headerEnricher", enricher, Message.class, Message.class);
|
InputEnricher enricher = new InputEnricher(configuration.getInputHeaderMappingExpression(), beanResolver);
|
||||||
composedFunction = (FunctionInvocationWrapper) w.andThen((Function<Object, Object>) composedFunction);
|
FunctionInvocationWrapper w = new FunctionInvocationWrapper("headerEnricher", enricher, Message.class, Message.class);
|
||||||
composedFunction.functionDefinition = functionDefinition;
|
composedFunction = (FunctionInvocationWrapper) w.andThen((Function<Object, Object>) composedFunction);
|
||||||
|
composedFunction.functionDefinition = functionDefinition;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
return composedFunction;
|
return composedFunction;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -22,7 +22,7 @@ import java.util.Date;
|
|||||||
import com.fasterxml.jackson.annotation.JsonFormat;
|
import com.fasterxml.jackson.annotation.JsonFormat;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* An example POJO that represents cloud event data
|
* An example POJO that represents cloud event data.
|
||||||
*
|
*
|
||||||
* @author Oleg Zhurakousky
|
* @author Oleg Zhurakousky
|
||||||
*
|
*
|
||||||
|
|||||||
@@ -16,7 +16,6 @@
|
|||||||
|
|
||||||
package org.springframework.cloud.function.context;
|
package org.springframework.cloud.function.context;
|
||||||
|
|
||||||
import java.util.Map;
|
|
||||||
import java.util.function.Function;
|
import java.util.function.Function;
|
||||||
|
|
||||||
import org.junit.jupiter.api.Test;
|
import org.junit.jupiter.api.Test;
|
||||||
@@ -24,13 +23,17 @@ import org.junit.jupiter.api.Test;
|
|||||||
import org.springframework.boot.WebApplicationType;
|
import org.springframework.boot.WebApplicationType;
|
||||||
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
|
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
|
||||||
import org.springframework.boot.builder.SpringApplicationBuilder;
|
import org.springframework.boot.builder.SpringApplicationBuilder;
|
||||||
import org.springframework.cloud.function.context.FunctionProperties.FunctionConfigurationProperties;
|
import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
|
||||||
import org.springframework.context.ConfigurableApplicationContext;
|
import org.springframework.context.ConfigurableApplicationContext;
|
||||||
import org.springframework.context.annotation.Bean;
|
import org.springframework.context.annotation.Bean;
|
||||||
import org.springframework.context.annotation.Configuration;
|
import org.springframework.context.annotation.Configuration;
|
||||||
|
import org.springframework.messaging.Message;
|
||||||
|
import org.springframework.messaging.MessageHeaders;
|
||||||
|
import org.springframework.messaging.support.MessageBuilder;
|
||||||
|
|
||||||
import static org.assertj.core.api.Assertions.assertThat;
|
import static org.assertj.core.api.Assertions.assertThat;
|
||||||
|
|
||||||
|
//NOTE!!! assertions for all tests are in 'echo' function since we're validating what's coming into it.
|
||||||
public class FunctionPropertiesTests {
|
public class FunctionPropertiesTests {
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
@@ -39,16 +42,14 @@ public class FunctionPropertiesTests {
|
|||||||
SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run(
|
SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run(
|
||||||
"--logging.level.org.springframework.cloud.function=DEBUG",
|
"--logging.level.org.springframework.cloud.function=DEBUG",
|
||||||
"--spring.main.lazy-initialization=true",
|
"--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].key1=hello1",
|
"--spring.cloud.function.configuration.echo.input-header-mapping-expression[0].key2='hello2'",
|
||||||
"--spring.cloud.function.configuration.echo.input-header-mapping-expression[0].key2=hello2",
|
"--spring.cloud.function.configuration.echo.input-header-mapping-expression[0].foo=headers.contentType")) {
|
||||||
"--spring.cloud.function.configuration.echo.input-header-mapping-expression[1].key12=hello12")) {
|
|
||||||
FunctionProperties functionProperties = context
|
FunctionCatalog functionCatalog = context.getBean(FunctionCatalog.class);
|
||||||
.getBean(FunctionProperties.class);
|
FunctionInvocationWrapper function = functionCatalog.lookup("echo");
|
||||||
FunctionConfigurationProperties configuration = functionProperties
|
function.apply(MessageBuilder.withPayload("helo")
|
||||||
.getConfiguration().get("echo");
|
.setHeader(MessageHeaders.CONTENT_TYPE, "application/json").build());
|
||||||
assertThat(configuration.getInputHeaderMappingExpression()).containsKey("0");
|
|
||||||
assertThat(configuration.getInputHeaderMappingExpression()).containsKey("1");
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -58,74 +59,86 @@ public class FunctionPropertiesTests {
|
|||||||
SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run(
|
SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run(
|
||||||
"--logging.level.org.springframework.cloud.function=DEBUG",
|
"--logging.level.org.springframework.cloud.function=DEBUG",
|
||||||
"--spring.main.lazy-initialization=true",
|
"--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.key1=hello1",
|
"--spring.cloud.function.configuration.echo.input-header-mapping-expression.key2='hello2'",
|
||||||
"--spring.cloud.function.configuration.echo.input-header-mapping-expression.key2=hello2")) {
|
"--spring.cloud.function.configuration.echo.input-header-mapping-expression.foo=headers.contentType")) {
|
||||||
FunctionProperties functionProperties = context
|
|
||||||
.getBean(FunctionProperties.class);
|
FunctionCatalog functionCatalog = context.getBean(FunctionCatalog.class);
|
||||||
FunctionConfigurationProperties configuration = functionProperties
|
FunctionInvocationWrapper function = functionCatalog.lookup("echo");
|
||||||
.getConfiguration().get("echo");
|
function.apply(MessageBuilder.withPayload("helo")
|
||||||
assertThat(configuration.getInputHeaderMappingExpression()).containsKey("0");
|
.setHeader(MessageHeaders.CONTENT_TYPE, "application/json").build());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
public void testInputHeaderMappingPropertyWithCompositionWithIndex() throws Exception {
|
public void testInputHeaderMappingExpressionWithCompositionWithIndex() throws Exception {
|
||||||
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
|
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
|
||||||
SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run(
|
SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run(
|
||||||
"--logging.level.org.springframework.cloud.function=DEBUG",
|
"--logging.level.org.springframework.cloud.function=DEBUG",
|
||||||
"--spring.main.lazy-initialization=true",
|
"--spring.main.lazy-initialization=true",
|
||||||
|
"--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[0].foo=headers.contentType")) {
|
||||||
|
|
||||||
|
FunctionCatalog functionCatalog = context.getBean(FunctionCatalog.class);
|
||||||
|
FunctionInvocationWrapper function = functionCatalog.lookup("echo|foo");
|
||||||
|
function.apply(MessageBuilder.withPayload("helo")
|
||||||
|
.setHeader(MessageHeaders.CONTENT_TYPE, "application/json").build());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void testInputHeaderMappingExpressionWithCompositionWithoutIndex() 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.configuration.echofoo.input-header-mapping-expression.key1='hello1'",
|
||||||
|
"--spring.cloud.function.configuration.echofoo.input-header-mapping-expression.key2='hello2'",
|
||||||
|
"--spring.cloud.function.configuration.echofoo.input-header-mapping-expression.foo=headers.contentType")) {
|
||||||
|
|
||||||
|
FunctionCatalog functionCatalog = context.getBean(FunctionCatalog.class);
|
||||||
|
FunctionInvocationWrapper function = functionCatalog.lookup("echo|foo");
|
||||||
|
function.apply(MessageBuilder.withPayload("helo")
|
||||||
|
.setHeader(MessageHeaders.CONTENT_TYPE, "application/json").build());
|
||||||
|
|
||||||
|
//assertions are in 'echo' function since we're validating what's coming into it.
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// @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.definition=echo|foo",
|
||||||
"--spring.cloud.function.configuration.echofoo.input-header-mapping-expression[0].key1=hello1",
|
// "--spring.cloud.function.configuration.echofoo.input-header-mapping-expression.key1=hello1",
|
||||||
"--spring.cloud.function.configuration.echofoo.input-header-mapping-expression[0].key2=hello2",
|
// "--spring.cloud.function.configuration.echofoo.input-header-mapping-expression.key2=hello2")) {
|
||||||
"--spring.cloud.function.configuration.echofoo.input-header-mapping-expression[1].key12=hello12")) {
|
// FunctionProperties functionProperties = context
|
||||||
FunctionProperties functionProperties = context
|
// .getBean(FunctionProperties.class);
|
||||||
.getBean(FunctionProperties.class);
|
// FunctionConfigurationProperties configuration = functionProperties
|
||||||
FunctionConfigurationProperties configuration = functionProperties
|
// .getConfiguration().get("echofoo");
|
||||||
.getConfiguration().get("echofoo");
|
|
||||||
Map<Object, Object> keyValueExpression = (Map<Object, Object>) configuration.getInputHeaderMappingExpression().get("0");
|
|
||||||
|
|
||||||
|
|
||||||
// FunctionCatalog functionCatalog = context.getBean(FunctionCatalog.class);
|
|
||||||
// FunctionInvocationWrapper function = functionCatalog.lookup("echo|foo");
|
|
||||||
// System.out.println(function.apply(new GenericMessage<String>("helo")));
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
// System.out.println(keyValueExpression.get("key1"));
|
|
||||||
// assertThat(configuration.getInputHeaderMappingExpression()).containsKey("0");
|
// 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
|
@EnableAutoConfiguration
|
||||||
@Configuration
|
@Configuration
|
||||||
protected static class SampleFunctionConfiguration {
|
protected static class SampleFunctionConfiguration {
|
||||||
|
|
||||||
@Bean
|
@Bean
|
||||||
public Function<String, String> echo() {
|
public Function<Message<?>, Message<?>> echo() {
|
||||||
return x -> x;
|
return m -> {
|
||||||
|
assertThat(m.getHeaders().get("key1")).isEqualTo("hello1");
|
||||||
|
assertThat(m.getHeaders().get("key2")).isEqualTo("hello2");
|
||||||
|
assertThat(m.getHeaders().get("foo")).isEqualTo("application/json");
|
||||||
|
return m;
|
||||||
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
@Bean
|
@Bean
|
||||||
public Function<String, String> foo() {
|
public Function<Message<?>, Message<?>> foo() {
|
||||||
return x -> x;
|
return x -> x;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user