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 1496f6c37d..9d2cbf00c2 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 @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-2018 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. @@ -56,10 +56,6 @@ public abstract class ConsumerEndpointSpec, protected ConsumerEndpointSpec(H messageHandler) { super(messageHandler); - this.endpointFactoryBean.setAdviceChain(this.adviceChain); - if (messageHandler instanceof AbstractReplyProducingMessageHandler) { - ((AbstractReplyProducingMessageHandler) messageHandler).setAdviceChain(this.adviceChain); - } } @Override @@ -270,6 +266,10 @@ public abstract class ConsumerEndpointSpec, @Override protected Tuple2 doGet() { + this.endpointFactoryBean.setAdviceChain(this.adviceChain); + if (this.handler instanceof AbstractReplyProducingMessageHandler) { + ((AbstractReplyProducingMessageHandler) this.handler).setAdviceChain(this.adviceChain); + } 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 2b5c822777..48cf0c04b8 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 @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-2018 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. @@ -55,14 +55,11 @@ public class HeaderEnricherSpec extends ConsumerEndpointSpec> headerToAdd = new HashMap<>(); - private boolean defaultOverwrite = false; - - private boolean shouldSkipNulls = true; - - private MessageProcessor messageProcessor; + private final HeaderEnricher headerEnricher = new HeaderEnricher(this.headerToAdd); HeaderEnricherSpec() { super(null); + this.handler = new MessageTransformingHandler(this.headerEnricher); } /** @@ -73,7 +70,7 @@ public class HeaderEnricherSpec extends ConsumerEndpointSpec messageProcessor) { - this.messageProcessor = messageProcessor; + this.headerEnricher.setMessageProcessor(messageProcessor); return _this(); } @@ -110,7 +107,7 @@ public class HeaderEnricherSpec extends ConsumerEndpointSpec(PARSER.parseExpression(expression))); + return messageProcessor(new ExpressionEvaluatingMessageProcessor<>(expression)); } /** @@ -371,6 +368,7 @@ public class HeaderEnricherSpec extends ConsumerEndpointSpec HeaderEnricherSpec headerFunction(String name, Function, Object> function, Boolean overwrite) { + return headerExpression(name, new FunctionExpression<>(function), overwrite); } @@ -391,6 +389,7 @@ public class HeaderEnricherSpec extends ConsumerEndpointSpec HeaderEnricherSpec header(String headerName, HeaderValueMessageProcessor headerValueMessageProcessor) { + Assert.hasText(headerName, "'headerName' must not be empty"); this.headerToAdd.put(headerName, headerValueMessageProcessor); return _this(); @@ -432,14 +431,7 @@ public class HeaderEnricherSpec extends ConsumerEndpointSpec doGet() { - HeaderEnricher headerEnricher = new HeaderEnricher(new HashMap<>(this.headerToAdd)); - headerEnricher.setDefaultOverwrite(this.defaultOverwrite); - headerEnricher.setShouldSkipNulls(this.shouldSkipNulls); - headerEnricher.setMessageProcessor(this.messageProcessor); - - this.componentsToRegister.put(headerEnricher, null); - - this.handler = new MessageTransformingHandler(headerEnricher); + this.componentsToRegister.put(this.headerEnricher, null); return super.doGet(); } 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 da17e3b4d5..0207dbc195 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 @@ -51,6 +51,7 @@ import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.dsl.Transformers; import org.springframework.integration.dsl.channel.MessageChannels; +import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice; import org.springframework.integration.handler.advice.IdempotentReceiverInterceptor; import org.springframework.integration.selector.MetadataStoreSelector; import org.springframework.integration.support.MessageBuilder; @@ -91,7 +92,6 @@ public class TransformerTests { private PollableChannel enricherErrorChannel; - @Test public void testContentEnricher() { QueueChannel replyChannel = new QueueChannel(); @@ -218,6 +218,9 @@ public class TransformerTests { @Autowired private PollableChannel idempotentDiscardChannel; + @Autowired + private PollableChannel adviceChannel; + @Test public void transformWithHeader() { QueueChannel replyChannel = new QueueChannel(); @@ -240,6 +243,7 @@ public class TransformerTests { } assertNotNull(this.idempotentDiscardChannel.receive(10000)); + assertNotNull(this.adviceChannel.receive(10000)); } @Configuration @@ -328,7 +332,7 @@ public class TransformerTests { return f -> f .enrichHeaders(h -> h .header("Foo", "Bar") - .advice(idempotentReceiverInterceptor())) + .advice(idempotentReceiverInterceptor(), requestHandlerAdvice())) .transform(new PojoTransformer()); } @@ -346,6 +350,26 @@ public class TransformerTests { return idempotentReceiverInterceptor; } + @Bean + public AbstractRequestHandlerAdvice requestHandlerAdvice() { + return new AbstractRequestHandlerAdvice() { + + @Override + protected Object doInvoke(ExecutionCallback callback, Object target, Message message) + throws Exception { + + adviceChannel().send(message); + return callback.execute(); + } + + }; + } + + @Bean + public PollableChannel adviceChannel() { + return new QueueChannel(); + } + @Bean public PollableChannel subFlowTestReplyChannel() { return new QueueChannel();