From 78bab07874b5fc2ac7a4d39f913aa19360c58f01 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 21 Mar 2017 16:32:01 -0400 Subject: [PATCH] Java DSL polishing around `HeaderEnricherSpec` * 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`) --- .../integration/dsl/BarrierSpec.java | 19 ++++----- .../integration/dsl/ConsumerEndpointSpec.java | 11 +++-- .../integration/dsl/HeaderEnricherSpec.java | 27 ++++++++---- .../dsl/IntegrationFlowDefinition.java | 41 ++++--------------- .../correlation/CorrelationHandlerTests.java | 2 +- .../dsl/transformers/TransformerTests.java | 5 ++- 6 files changed, 47 insertions(+), 58 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/BarrierSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/BarrierSpec.java index c03738c739..8c48a04ec0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/BarrierSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/BarrierSpec.java @@ -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 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(); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java index d894706608..91107c7456 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java @@ -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, 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, return _this(); } + @Override + protected Tuple2 doGet() { + this.endpointFactoryBean.setHandler(this.handler); + return super.doGet(); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/HeaderEnricherSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/HeaderEnricherSpec.java index 6a14d572c0..dbe82e18db 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/HeaderEnricherSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/HeaderEnricherSpec.java @@ -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 { +public class HeaderEnricherSpec extends ConsumerEndpointSpec { private final Map> headerToAdd = new HashMap<>(); @@ -58,6 +62,7 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec messageProcessor; HeaderEnricherSpec() { + super(null); } /** @@ -105,7 +110,7 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec(PARSER.parseExpression(expression))); + return messageProcessor(new ExpressionEvaluatingMessageProcessor<>(PARSER.parseExpression(expression))); } /** @@ -120,7 +125,7 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec(beanName, methodName)); + return messageProcessor(new BeanNameMessageProcessor<>(beanName, methodName)); } /** @@ -177,7 +182,7 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec processor = - new ExpressionEvaluatingHeaderValueMessageProcessor((Expression) value, null); + new ExpressionEvaluatingHeaderValueMessageProcessor<>((Expression) value, null); processor.setOverwrite(overwrite); header(name, processor); } @@ -281,7 +286,7 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec entry : headers.entrySet()) { AbstractHeaderValueMessageProcessor processor = - new ExpressionEvaluatingHeaderValueMessageProcessor(entry.getValue(), null); + new ExpressionEvaluatingHeaderValueMessageProcessor<>(entry.getValue(), null); processor.setOverwrite(overwrite); header(entry.getKey(), processor); } @@ -310,7 +315,7 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec HeaderEnricherSpec header(String name, V value, Boolean overwrite) { AbstractHeaderValueMessageProcessor headerValueMessageProcessor = - new StaticHeaderValueMessageProcessor(value); + new StaticHeaderValueMessageProcessor<>(value); headerValueMessageProcessor.setOverwrite(overwrite); return header(name, headerValueMessageProcessor); } @@ -371,7 +376,7 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec headerValueMessageProcessor = - new ExpressionEvaluatingHeaderValueMessageProcessor(expression, null); + new ExpressionEvaluatingHeaderValueMessageProcessor<>(expression, null); headerValueMessageProcessor.setOverwrite(overwrite); return header(name, headerValueMessageProcessor); } @@ -426,12 +431,16 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec 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(); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java index daa9006a7f..e86c5954e9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java @@ -1244,7 +1244,11 @@ public abstract class IntegrationFlowDefinition headers, Consumer> endpointConfigurer) { - return enrichHeaders(spec -> spec.headers(headers), endpointConfigurer); + HeaderEnricherSpec headerEnricherSpec = new HeaderEnricherSpec(); + headerEnricherSpec.headers(headers); + Tuple2 tuple2 = headerEnricherSpec.get(); + return addComponents(headerEnricherSpec.getComponentsToRegister()) + .handle(tuple2.getT2(), endpointConfigurer); } /** @@ -1263,36 +1267,8 @@ public abstract class IntegrationFlowDefinition 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: - * 
-	 * {@code
-	 *  .enrichHeaders(
-	 *s -> s.header("one", new XPathExpressionEvaluatingHeaderValueMessageProcessor("/root/elementOne"))
-	 *            .header("two", new XPathExpressionEvaluatingHeaderValueMessageProcessor("/root/elementTwo"))
-	 *            .headerChannelsToString(),
-	 *            c -> c.autoStartup(false).id("xpathHeaderEnricher"))
-	 * }
-	 *
- * @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 headerEnricherConfigurer, - Consumer> 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 factoryBeanTuple2 = endpointSpec.get(); + + addComponents(endpointSpec.getComponentsToRegister()); + if (inputChannel instanceof MessageChannelReference) { factoryBeanTuple2.getT1().setInputChannelName(((MessageChannelReference) inputChannel).getName()); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java index 35e1e2693f..f57483c12b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java @@ -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) diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/transformers/TransformerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/transformers/TransformerTests.java index dc104075e0..1c15f49372 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/transformers/TransformerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/transformers/TransformerTests.java @@ -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()); }