Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Java DSL polishing around HeaderEnricherSpec

Browse Source 
* Make `HeaderEnricherSpec extends ConsumerEndpointSpec` to avoid extra
EIP-method with two `Consumer`s, when only one for
the `HeaderEnricherSpec` can address all the options - header enricher,
as well as target endpoint
* Move `this.endpointFactoryBean.setHandler(this.handler)` into
the `ConsumerEndpointSpec#doGet()` instead of ctor to avoid
duplicate code from the target implementations (e.g. `BarrierSpec`)
This commit is contained in:
Artem Bilan
2017-03-21 16:32:01 -04:00
committed by Gary Russell
parent 515042fa03
commit 78bab07874
6 changed files with 47 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
spring-integration-core/src/main/java/org/springframework/integration/dsl/BarrierSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2017 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.
@@ -27,7 +27,6 @@ import org.springframework.integration.config.ConsumerEndpointFactoryBean;
import org.springframework.util.Assert;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuples;
/**
* A {@link MessageHandlerSpec} for the {@link BarrierMessageHandler}.
@@ -96,15 +95,13 @@ public class BarrierSpec extends ConsumerEndpointSpec<BarrierSpec, BarrierMessag
@Override
public Tuple2<ConsumerEndpointFactoryBean, BarrierMessageHandler> doGet() {
BarrierMessageHandler barrierMessageHandler =
new BarrierMessageHandler(this.timeout, this.outputProcessor, this.correlationStrategy);
barrierMessageHandler.setAdviceChain(this.adviceChain);
barrierMessageHandler.setRequiresReply(this.requiresReply);
barrierMessageHandler.setSendTimeout(this.sendTimeout);
barrierMessageHandler.setAsync(this.async);
barrierMessageHandler.setOrder(this.order);
this.endpointFactoryBean.setHandler(barrierMessageHandler);
return Tuples.of(this.endpointFactoryBean, barrierMessageHandler);
this.handler = new BarrierMessageHandler(this.timeout, this.outputProcessor, this.correlationStrategy);
this.handler.setAdviceChain(this.adviceChain);
this.handler.setRequiresReply(this.requiresReply);
this.handler.setSendTimeout(this.sendTimeout);
this.handler.setAsync(this.async);
this.handler.setOrder(this.order);
return super.doGet();
}
}

11
spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java
View File

@@ -34,6 +34,8 @@ import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
import org.springframework.transaction.interceptor.TransactionInterceptor;
import reactor.util.function.Tuple2;
/**
* A {@link EndpointSpec} for consumer endpoints.
*
@@ -52,9 +54,6 @@ public abstract class ConsumerEndpointSpec<S extends ConsumerEndpointSpec<S, H>,
protected ConsumerEndpointSpec(H messageHandler) {
super(messageHandler);
if (messageHandler != null) {
this.endpointFactoryBean.setHandler(messageHandler);
}
this.endpointFactoryBean.setAdviceChain(this.adviceChain);
if (messageHandler instanceof AbstractReplyProducingMessageHandler) {
((AbstractReplyProducingMessageHandler) messageHandler).setAdviceChain(this.adviceChain);
@@ -227,4 +226,10 @@ public abstract class ConsumerEndpointSpec<S extends ConsumerEndpointSpec<S, H>,
return _this();
}
@Override
protected Tuple2<ConsumerEndpointFactoryBean, H> doGet() {
this.endpointFactoryBean.setHandler(this.handler);
return super.doGet();
}
}

27
spring-integration-core/src/main/java/org/springframework/integration/dsl/HeaderEnricherSpec.java
View File

@@ -23,6 +23,7 @@ import java.util.function.Consumer;
import java.util.function.Function;
import org.springframework.expression.Expression;
import org.springframework.integration.config.ConsumerEndpointFactoryBean;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.expression.FunctionExpression;
import org.springframework.integration.handler.BeanNameMessageProcessor;
@@ -31,6 +32,7 @@ import org.springframework.integration.handler.MessageProcessor;
import org.springframework.integration.support.MapBuilder;
import org.springframework.integration.support.StringStringMapBuilder;
import org.springframework.integration.transformer.HeaderEnricher;
import org.springframework.integration.transformer.MessageTransformingHandler;
import org.springframework.integration.transformer.support.AbstractHeaderValueMessageProcessor;
import org.springframework.integration.transformer.support.ExpressionEvaluatingHeaderValueMessageProcessor;
import org.springframework.integration.transformer.support.HeaderValueMessageProcessor;
@@ -39,6 +41,8 @@ import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import reactor.util.function.Tuple2;
/**
* An {@link IntegrationComponentSpec} for a {@link HeaderEnricher}.
*
@@ -47,7 +51,7 @@ import org.springframework.util.StringUtils;
*
* @since 5.0
*/
public class HeaderEnricherSpec extends IntegrationComponentSpec<HeaderEnricherSpec, HeaderEnricher> {
public class HeaderEnricherSpec extends ConsumerEndpointSpec<HeaderEnricherSpec, MessageTransformingHandler> {
private final Map<String, HeaderValueMessageProcessor<?>> headerToAdd = new HashMap<>();
@@ -58,6 +62,7 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec<HeaderEnricherS
private MessageProcessor<?> messageProcessor;
HeaderEnricherSpec() {
super(null);
}
/**
@@ -105,7 +110,7 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec<HeaderEnricherS
* @see #messageProcessor(MessageProcessor)
*/
public HeaderEnricherSpec messageProcessor(String expression) {
return messageProcessor(new ExpressionEvaluatingMessageProcessor<Object>(PARSER.parseExpression(expression)));
return messageProcessor(new ExpressionEvaluatingMessageProcessor<>(PARSER.parseExpression(expression)));
}
/**
@@ -120,7 +125,7 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec<HeaderEnricherS
* @see #messageProcessor(MessageProcessor)
*/
public HeaderEnricherSpec messageProcessor(String beanName, String methodName) {
return messageProcessor(new BeanNameMessageProcessor<Object>(beanName, methodName));
return messageProcessor(new BeanNameMessageProcessor<>(beanName, methodName));
}
/**
@@ -177,7 +182,7 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec<HeaderEnricherS
Object value = entry.getValue();
if (value instanceof Expression) {
AbstractHeaderValueMessageProcessor<Object> processor =
new ExpressionEvaluatingHeaderValueMessageProcessor<Object>((Expression) value, null);
new ExpressionEvaluatingHeaderValueMessageProcessor<>((Expression) value, null);
processor.setOverwrite(overwrite);
header(name, processor);
}
@@ -281,7 +286,7 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec<HeaderEnricherS
Assert.notNull(headers, "'headers' must not be null");
for (Entry<String, String> entry : headers.entrySet()) {
AbstractHeaderValueMessageProcessor<Object> processor =
new ExpressionEvaluatingHeaderValueMessageProcessor<Object>(entry.getValue(), null);
new ExpressionEvaluatingHeaderValueMessageProcessor<>(entry.getValue(), null);
processor.setOverwrite(overwrite);
header(entry.getKey(), processor);
}
@@ -310,7 +315,7 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec<HeaderEnricherS
*/
public <V> HeaderEnricherSpec header(String name, V value, Boolean overwrite) {
AbstractHeaderValueMessageProcessor<V> headerValueMessageProcessor =
new StaticHeaderValueMessageProcessor<V>(value);
new StaticHeaderValueMessageProcessor<>(value);
headerValueMessageProcessor.setOverwrite(overwrite);
return header(name, headerValueMessageProcessor);
}
@@ -371,7 +376,7 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec<HeaderEnricherS
private HeaderEnricherSpec headerExpression(String name, Expression expression, Boolean overwrite) {
AbstractHeaderValueMessageProcessor<?> headerValueMessageProcessor =
new ExpressionEvaluatingHeaderValueMessageProcessor<Object>(expression, null);
new ExpressionEvaluatingHeaderValueMessageProcessor<>(expression, null);
headerValueMessageProcessor.setOverwrite(overwrite);
return header(name, headerValueMessageProcessor);
}
@@ -426,12 +431,16 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec<HeaderEnricherS
}
@Override
protected HeaderEnricher doGet() {
protected Tuple2<ConsumerEndpointFactoryBean, MessageTransformingHandler> doGet() {
HeaderEnricher headerEnricher = new HeaderEnricher(new HashMap<>(this.headerToAdd));
headerEnricher.setDefaultOverwrite(this.defaultOverwrite);
headerEnricher.setShouldSkipNulls(this.shouldSkipNulls);
headerEnricher.setMessageProcessor(this.messageProcessor);
return headerEnricher;
this.componentsToRegister.add(headerEnricher);
this.handler = new MessageTransformingHandler(headerEnricher);
return super.doGet();
}
}

41
spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java
View File

@@ -1244,7 +1244,11 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
*/
public B enrichHeaders(final Map<String, Object> headers,
Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
return enrichHeaders(spec -> spec.headers(headers), endpointConfigurer);
HeaderEnricherSpec headerEnricherSpec = new HeaderEnricherSpec();
headerEnricherSpec.headers(headers);
Tuple2<ConsumerEndpointFactoryBean, MessageTransformingHandler> tuple2 = headerEnricherSpec.get();
return addComponents(headerEnricherSpec.getComponentsToRegister())
.handle(tuple2.getT2(), endpointConfigurer);
}
/**
@@ -1263,36 +1267,8 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @see HeaderEnricherSpec
*/
public B enrichHeaders(Consumer<HeaderEnricherSpec> headerEnricherConfigurer) {
return this.enrichHeaders(headerEnricherConfigurer, null);
}
/**
* Populate a {@link MessageTransformingHandler} for
* a {@link org.springframework.integration.transformer.HeaderEnricher}
* as the result of provided {@link Consumer}.
* In addition accept options for the integration endpoint using {@link GenericEndpointSpec}.
* Typically used with a Java 8 Lambda expression:
* <pre class="code">
* {@code
* .enrichHeaders(
*s -> s.header("one", new XPathExpressionEvaluatingHeaderValueMessageProcessor("/root/elementOne"))
* .header("two", new XPathExpressionEvaluatingHeaderValueMessageProcessor("/root/elementTwo"))
* .headerChannelsToString(),
* c -> c.autoStartup(false).id("xpathHeaderEnricher"))
* }
* </pre>
* @param headerEnricherConfigurer the {@link Consumer} to use.
* @param endpointConfigurer the {@link Consumer} to provide integration endpoint options.
* @return the current {@link IntegrationFlowDefinition}.
* @see HeaderEnricherSpec
* @see GenericEndpointSpec
*/
public B enrichHeaders(Consumer<HeaderEnricherSpec> headerEnricherConfigurer,
Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
Assert.notNull(headerEnricherConfigurer, "'headerEnricherConfigurer' must not be null");
HeaderEnricherSpec headerEnricherSpec = new HeaderEnricherSpec();
headerEnricherConfigurer.accept(headerEnricherSpec);
return transform(headerEnricherSpec.get(), endpointConfigurer);
return register(new HeaderEnricherSpec(), headerEnricherConfigurer);
}
/**
@@ -2591,8 +2567,6 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
endpointConfigurer.accept(endpointSpec);
}
addComponents(endpointSpec.getComponentsToRegister());
MessageChannel inputChannel = this.currentMessageChannel;
this.currentMessageChannel = null;
if (inputChannel == null) {
@@ -2601,6 +2575,9 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
}
Tuple2<ConsumerEndpointFactoryBean, ? extends MessageHandler> factoryBeanTuple2 = endpointSpec.get();
addComponents(endpointSpec.getComponentsToRegister());
if (inputChannel instanceof MessageChannelReference) {
factoryBeanTuple2.getT1().setInputChannelName(((MessageChannelReference) inputChannel).getName());
}

2
spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java
View File

@@ -177,7 +177,7 @@ public class CorrelationHandlerTests {
return f -> f.enrichHeaders(s -> s.header("FOO", "BAR"))
.split("testSplitterData", "buildList", c -> c.applySequence(false))
.channel(MessageChannels.executor(taskExecutor()))
.split(Message.class, m -> m.getPayload(), c -> c.applySequence(false))
.split(Message.class, Message::getPayload, c -> c.applySequence(false))
.channel(MessageChannels.executor(taskExecutor()))
.split(s -> s
.applySequence(false)

5
spring-integration-core/src/test/java/org/springframework/integration/dsl/transformers/TransformerTests.java
View File

@@ -304,8 +304,9 @@ public class TransformerTests {
@Bean
public IntegrationFlow pojoTransformFlow() {
return f -> f
.enrichHeaders(h -> h.header("Foo", "Bar"),
e -> e.advice(idempotentReceiverInterceptor()))
.enrichHeaders(h -> h
.header("Foo", "Bar")
.advice(idempotentReceiverInterceptor()))
.transform(new PojoTransformer());
}