@@ -596,7 +596,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
|
||||
if (result instanceof Message) {
|
||||
Map<String, Object> headersMap = (Map<String, Object>) ReflectionUtils
|
||||
.getField(SimpleFunctionRegistry.this.headersField, ((Message) result).getHeaders());
|
||||
headersMap.putAll(this.sanitizeHeaders(((Message) input).getHeaders()));
|
||||
this.sanitizeHeaders(((Message) input).getHeaders()).forEach((k, v) -> headersMap.putIfAbsent(k, v));
|
||||
}
|
||||
else {
|
||||
result = MessageBuilder.withPayload(result).copyHeaders(this.sanitizeHeaders(((Message) input).getHeaders())).build();
|
||||
|
||||
@@ -311,6 +311,7 @@ public class SimpleFunctionRegistryTests {
|
||||
assertThat(function).isNotNull();
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
@Test
|
||||
public void lookupWithCompositionFunctionAndConsumer() {
|
||||
SimpleFunctionRegistry functionRegistry = new SimpleFunctionRegistry(this.conversionService, this.messageConverter,
|
||||
@@ -347,6 +348,22 @@ public class SimpleFunctionRegistryTests {
|
||||
functionWrapper.apply("123");
|
||||
}
|
||||
|
||||
@SuppressWarnings({ "rawtypes", "unchecked" })
|
||||
@Test
|
||||
public void testHeaderEnricherFunction() {
|
||||
FunctionRegistration<HeaderEnricherFunction> registration =
|
||||
new FunctionRegistration<>(new HeaderEnricherFunction(), "headerEnricher")
|
||||
.type(FunctionType.of(HeaderEnricherFunction.class));
|
||||
SimpleFunctionRegistry catalog = new SimpleFunctionRegistry(this.conversionService, this.messageConverter,
|
||||
new JacksonMapper(new ObjectMapper()));
|
||||
catalog.register(registration);
|
||||
Function<Message<?>, Message<?>> function = catalog.lookup("headerEnricher");
|
||||
Message<?> message =
|
||||
function.apply(MessageBuilder.withPayload("hello").setHeader("original", "originalValue")
|
||||
.build());
|
||||
assertThat(message.getHeaders().get("original")).isEqualTo("newValue");
|
||||
}
|
||||
|
||||
|
||||
public Function<String, String> uppercase() {
|
||||
return v -> v.toUpperCase();
|
||||
@@ -501,4 +518,13 @@ public class SimpleFunctionRegistryTests {
|
||||
.map(lst -> lst.stream().map(Person::getName).collect(Collectors.toList()));
|
||||
}
|
||||
}
|
||||
|
||||
private static class HeaderEnricherFunction implements Function<Message<?>, Message<?>> {
|
||||
|
||||
@Override
|
||||
public Message<?> apply(Message<?> message) {
|
||||
return MessageBuilder.withPayload(message.getPayload()).setHeader("original", "newValue")
|
||||
.build();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -22,6 +22,7 @@ import java.util.function.Function;
|
||||
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import reactor.core.publisher.Flux;
|
||||
|
||||
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
|
||||
import org.springframework.boot.builder.SpringApplicationBuilder;
|
||||
@@ -34,7 +35,6 @@ import org.springframework.core.ResolvableType;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
|
||||
import reactor.core.publisher.Flux;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
@@ -104,9 +104,9 @@ public class UserIssuesTests {
|
||||
FunctionCatalog catalog = this.configureCatalog(Issue601Configuration.class);
|
||||
FunctionInvocationWrapper function = catalog.lookup("uppercase");
|
||||
assertThat(function.getInputType().getTypeName())
|
||||
.isEqualTo(ResolvableType.forClassWithGenerics(Flux.class, String.class).getType().getTypeName());
|
||||
.isEqualTo(ResolvableType.forClassWithGenerics(Flux.class, String.class).getType().getTypeName());
|
||||
assertThat(function.getOutputType().getTypeName())
|
||||
.isEqualTo(ResolvableType.forClassWithGenerics(Flux.class, Integer.class).getType().getTypeName());
|
||||
.isEqualTo(ResolvableType.forClassWithGenerics(Flux.class, Integer.class).getType().getTypeName());
|
||||
Flux<Integer> result = (Flux<Integer>) function.apply(Flux.just("julien", "ricky", "bubbles"));
|
||||
List<Integer> results = result.collectList().block();
|
||||
assertThat(results.get(0)).isEqualTo(6);
|
||||
@@ -114,7 +114,6 @@ public class UserIssuesTests {
|
||||
assertThat(results.get(2)).isEqualTo(7);
|
||||
}
|
||||
|
||||
|
||||
@EnableAutoConfiguration
|
||||
@Configuration
|
||||
public static class Issue602Configuration {
|
||||
@@ -133,17 +132,17 @@ public class UserIssuesTests {
|
||||
@Configuration
|
||||
public static class Issue601Configuration {
|
||||
@Bean
|
||||
public Uppercase uppercase() {
|
||||
return new Uppercase();
|
||||
}
|
||||
public Uppercase uppercase() {
|
||||
return new Uppercase();
|
||||
}
|
||||
}
|
||||
|
||||
public static class Uppercase implements Function<Flux<String>, Flux<Integer>> {
|
||||
|
||||
@Override
|
||||
public Flux<Integer> apply(Flux<String> s) {
|
||||
return s.map(v -> v.length());
|
||||
}
|
||||
@Override
|
||||
public Flux<Integer> apply(Flux<String> s) {
|
||||
return s.map(v -> v.length());
|
||||
}
|
||||
}
|
||||
|
||||
public static class Product {
|
||||
|
||||
Reference in New Issue
Block a user