diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java index 516882be63..f3883ca7c8 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java @@ -29,14 +29,16 @@ import java.util.List; import java.util.concurrent.atomic.AtomicBoolean; import org.junit.Test; +import org.junit.runner.RunWith; import org.mockito.Mockito; import org.springframework.amqp.core.MessageDeliveryMode; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.support.AmqpHeaders; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.parsing.BeanDefinitionParsingException; -import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint; import org.springframework.integration.amqp.outbound.AsyncAmqpOutboundGateway; @@ -47,6 +49,8 @@ import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; import org.springframework.util.ReflectionUtils; /** @@ -58,31 +62,32 @@ import org.springframework.util.ReflectionUtils; * @since 2.1 * */ +@RunWith(SpringRunner.class) +@DirtiesContext public class AmqpOutboundGatewayParserTests { private static volatile int adviceCalled; + @Autowired + private ApplicationContext context; + @Test public void testGatewayConfig() { - ConfigurableApplicationContext context = new ClassPathXmlApplicationContext( - "AmqpOutboundGatewayParserTests-context.xml", this.getClass()); - Object edc = context.getBean("rabbitGateway"); + Object edc = this.context.getBean("rabbitGateway"); assertFalse(TestUtils.getPropertyValue(edc, "autoStartup", Boolean.class)); AmqpOutboundEndpoint gateway = TestUtils.getPropertyValue(edc, "handler", AmqpOutboundEndpoint.class); assertEquals("amqp:outbound-gateway", gateway.getComponentType()); assertTrue(TestUtils.getPropertyValue(gateway, "requiresReply", Boolean.class)); - checkGWProps(context, gateway); + checkGWProps(this.context, gateway); - AsyncAmqpOutboundGateway async = context.getBean("asyncGateway.handler", AsyncAmqpOutboundGateway.class); + AsyncAmqpOutboundGateway async = this.context.getBean("asyncGateway.handler", AsyncAmqpOutboundGateway.class); assertEquals("amqp:outbound-async-gateway", async.getComponentType()); - checkGWProps(context, async); - assertSame(context.getBean("asyncTemplate"), TestUtils.getPropertyValue(async, "template")); - assertSame(context.getBean("ems"), TestUtils.getPropertyValue(gateway, "errorMessageStrategy")); - - context.close(); + checkGWProps(this.context, async); + assertSame(this.context.getBean("asyncTemplate"), TestUtils.getPropertyValue(async, "template")); + assertSame(this.context.getBean("ems"), TestUtils.getPropertyValue(gateway, "errorMessageStrategy")); } - protected void checkGWProps(ConfigurableApplicationContext context, Orderable gateway) { + protected void checkGWProps(ApplicationContext context, Orderable gateway) { assertEquals(5, gateway.getOrder()); assertEquals(context.getBean("fromRabbit"), TestUtils.getPropertyValue(gateway, "outputChannel")); MessageChannel returnChannel = context.getBean("returnChannel", MessageChannel.class); @@ -97,12 +102,9 @@ public class AmqpOutboundGatewayParserTests { .getExpressionString()); } - @SuppressWarnings({ "resource" }) @Test public void withHeaderMapperCustomRequestResponse() { - ConfigurableApplicationContext context = new ClassPathXmlApplicationContext( - "AmqpOutboundGatewayParserTests-context.xml", this.getClass()); - Object eventDrivenConsumer = context.getBean("withHeaderMapperCustomRequestResponse"); + Object eventDrivenConsumer = this.context.getBean("withHeaderMapperCustomRequestResponse"); AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivenConsumer, "handler", AmqpOutboundEndpoint.class); @@ -136,7 +138,7 @@ public class AmqpOutboundGatewayParserTests { Mockito.any(org.springframework.amqp.core.Message.class), isNull()); ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate); - MessageChannel requestChannel = context.getBean("toRabbit1", MessageChannel.class); + MessageChannel requestChannel = this.context.getBean("toRabbit1", MessageChannel.class); Message message = MessageBuilder.withPayload("hello").setHeader("foo", "foo").build(); requestChannel.send(message); @@ -145,7 +147,7 @@ public class AmqpOutboundGatewayParserTests { isNull()); // verify reply - QueueChannel queueChannel = context.getBean("fromRabbit", QueueChannel.class); + QueueChannel queueChannel = this.context.getBean("fromRabbit", QueueChannel.class); Message replyMessage = queueChannel.receive(0); assertNotNull(replyMessage); assertEquals("bar", replyMessage.getHeaders().get("bar")); @@ -163,16 +165,11 @@ public class AmqpOutboundGatewayParserTests { requestChannel.send(message); replyMessage = queueChannel.receive(0); assertNotNull(replyMessage); - - context.close(); } - @SuppressWarnings({ "resource" }) @Test public void withHeaderMapperCustomAndStandardResponse() { - ConfigurableApplicationContext context = new ClassPathXmlApplicationContext( - "AmqpOutboundGatewayParserTests-context.xml", this.getClass()); - Object eventDrivenConsumer = context.getBean("withHeaderMapperCustomAndStandardResponse"); + Object eventDrivenConsumer = this.context.getBean("withHeaderMapperCustomAndStandardResponse"); AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivenConsumer, "handler", AmqpOutboundEndpoint.class); @@ -202,7 +199,7 @@ public class AmqpOutboundGatewayParserTests { Mockito.any(org.springframework.amqp.core.Message.class), isNull()); ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate); - MessageChannel requestChannel = context.getBean("toRabbit2", MessageChannel.class); + MessageChannel requestChannel = this.context.getBean("toRabbit2", MessageChannel.class); Message message = MessageBuilder.withPayload("hello").setHeader("foo", "foo").build(); requestChannel.send(message); @@ -211,7 +208,7 @@ public class AmqpOutboundGatewayParserTests { isNull()); // verify reply - QueueChannel queueChannel = context.getBean("fromRabbit", QueueChannel.class); + QueueChannel queueChannel = this.context.getBean("fromRabbit", QueueChannel.class); Message replyMessage = queueChannel.receive(0); assertEquals("bar", replyMessage.getHeaders().get("bar")); assertEquals("foo", replyMessage.getHeaders().get("foo")); // copied from request Message @@ -219,15 +216,11 @@ public class AmqpOutboundGatewayParserTests { assertNotNull(replyMessage.getHeaders().get(AmqpHeaders.RECEIVED_DELIVERY_MODE)); assertNotNull(replyMessage.getHeaders().get(AmqpHeaders.CONTENT_TYPE)); assertNotNull(replyMessage.getHeaders().get(AmqpHeaders.APP_ID)); - context.close(); } - @SuppressWarnings({ "resource" }) @Test public void withHeaderMapperNothingToMap() { - ConfigurableApplicationContext context = new ClassPathXmlApplicationContext( - "AmqpOutboundGatewayParserTests-context.xml", this.getClass()); - Object eventDrivenConsumer = context.getBean("withHeaderMapperNothingToMap"); + Object eventDrivenConsumer = this.context.getBean("withHeaderMapperNothingToMap"); AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivenConsumer, "handler", AmqpOutboundEndpoint.class); @@ -253,7 +246,7 @@ public class AmqpOutboundGatewayParserTests { Mockito.any(org.springframework.amqp.core.Message.class), isNull()); ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate); - MessageChannel requestChannel = context.getBean("toRabbit3", MessageChannel.class); + MessageChannel requestChannel = this.context.getBean("toRabbit3", MessageChannel.class); Message message = MessageBuilder.withPayload("hello").setHeader("foo", "foo").build(); requestChannel.send(message); @@ -271,15 +264,11 @@ public class AmqpOutboundGatewayParserTests { assertNull(replyMessage.getHeaders().get(AmqpHeaders.CONTENT_TYPE)); assertNull(replyMessage.getHeaders().get(AmqpHeaders.APP_ID)); assertEquals(1, adviceCalled); - context.close(); } - @SuppressWarnings("resource") @Test //INT-1029 public void amqpOutboundGatewayWithinChain() { - ConfigurableApplicationContext context = new ClassPathXmlApplicationContext( - "AmqpOutboundGatewayParserTests-context.xml", this.getClass()); - Object eventDrivenConsumer = context.getBean("chainWithRabbitOutboundGateway"); + Object eventDrivenConsumer = this.context.getBean("chainWithRabbitOutboundGateway"); List chainHandlers = TestUtils.getPropertyValue(eventDrivenConsumer, "handler.handlers", List.class); @@ -307,7 +296,7 @@ public class AmqpOutboundGatewayParserTests { ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate); - MessageChannel requestChannel = context.getBean("toRabbit4", MessageChannel.class); + MessageChannel requestChannel = this.context.getBean("toRabbit4", MessageChannel.class); Message message = MessageBuilder.withPayload("hello").setHeader("foo", "foo").build(); requestChannel.send(message); @@ -316,7 +305,7 @@ public class AmqpOutboundGatewayParserTests { isNull()); // verify reply - QueueChannel queueChannel = context.getBean("fromRabbit", QueueChannel.class); + QueueChannel queueChannel = this.context.getBean("fromRabbit", QueueChannel.class); Message replyMessage = queueChannel.receive(0); assertEquals("hello", new String((byte[]) replyMessage.getPayload())); assertNull(replyMessage.getHeaders().get("bar")); @@ -325,7 +314,6 @@ public class AmqpOutboundGatewayParserTests { assertNull(replyMessage.getHeaders().get(AmqpHeaders.DELIVERY_MODE)); assertNull(replyMessage.getHeaders().get(AmqpHeaders.CONTENT_TYPE)); assertNull(replyMessage.getHeaders().get(AmqpHeaders.APP_ID)); - context.close(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractStandardMessageHandlerFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractStandardMessageHandlerFactoryBean.java index b1fbbe1a43..084a3d0572 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractStandardMessageHandlerFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractStandardMessageHandlerFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-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. @@ -23,7 +23,6 @@ import org.springframework.aop.TargetSource; import org.springframework.aop.framework.Advised; import org.springframework.expression.Expression; import org.springframework.expression.ExpressionParser; -import org.springframework.expression.spel.SpelParserConfiguration; import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.handler.AbstractMessageProducingHandler; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; @@ -44,10 +43,11 @@ import org.springframework.util.StringUtils; public abstract class AbstractStandardMessageHandlerFactoryBean extends AbstractSimpleMessageHandlerFactoryBean { - private static final ExpressionParser expressionParser = new SpelExpressionParser(new SpelParserConfiguration(true, - true)); + private static final ExpressionParser expressionParser = new SpelExpressionParser(); - private static final Set referencedReplyProducers = new HashSet(); + private static final Set referencedReplyProducers = new HashSet<>(); + + private volatile Boolean requiresReply; private volatile Object targetObject; @@ -55,6 +55,8 @@ public abstract class AbstractStandardMessageHandlerFactoryBean private volatile Expression expression; + private volatile Long sendTimeout; + /** * Set the target POJO for the message handler. * @param targetObject the target object. @@ -87,6 +89,18 @@ public abstract class AbstractStandardMessageHandlerFactoryBean this.expression = expression; } + public void setRequiresReply(Boolean requiresReply) { + this.requiresReply = requiresReply; + } + + public void setSendTimeout(Long sendTimeout) { + this.sendTimeout = sendTimeout; + } + + public Long getSendTimeout() { + return this.sendTimeout; + } + @Override protected MessageHandler createHandler() { MessageHandler handler; @@ -100,8 +114,8 @@ public abstract class AbstractStandardMessageHandlerFactoryBean AbstractMessageProducingHandler actualHandler = this.extractTypeIfPossible(this.targetObject, AbstractMessageProducingHandler.class); boolean targetIsDirectReplyProducingHandler = actualHandler != null - && this.canBeUsedDirect(actualHandler) // give subclasses a say - && this.methodIsHandleMessageOrEmpty(this.targetMethodName); + && canBeUsedDirect(actualHandler) // give subclasses a say + && methodIsHandleMessageOrEmpty(this.targetMethodName); if (this.targetObject instanceof MessageProcessor) { handler = this.createMessageProcessingHandler((MessageProcessor) this.targetObject); } @@ -109,8 +123,8 @@ public abstract class AbstractStandardMessageHandlerFactoryBean if (logger.isDebugEnabled()) { logger.debug("Wiring handler (" + this.targetObject + ") directly into endpoint"); } - this.checkReuse(actualHandler); - this.postProcessReplyProducer(actualHandler); + checkReuse(actualHandler); + postProcessReplyProducer(actualHandler); handler = (MessageHandler) this.targetObject; } else { @@ -142,7 +156,7 @@ public abstract class AbstractStandardMessageHandlerFactoryBean private void checkReuse(AbstractMessageProducingHandler replyHandler) { Assert.isTrue(!referencedReplyProducers.contains(replyHandler), "An AbstractMessageProducingMessageHandler may only be referenced once (" + - replyHandler.getComponentName() + ") - use scope=\"prototype\""); + replyHandler.getComponentName() + ") - use scope=\"prototype\""); referencedReplyProducers.add(replyHandler); } @@ -176,9 +190,6 @@ public abstract class AbstractStandardMessageHandlerFactoryBean } if (targetObject instanceof Advised) { TargetSource targetSource = ((Advised) targetObject).getTargetSource(); - if (targetSource == null) { - return null; - } try { return extractTypeIfPossible(targetSource.getTarget(), expectedType); } @@ -199,6 +210,21 @@ public abstract class AbstractStandardMessageHandlerFactoryBean } protected void postProcessReplyProducer(AbstractMessageProducingHandler handler) { + if (this.sendTimeout != null) { + handler.setSendTimeout(this.sendTimeout); + } + + if (this.requiresReply != null) { + if (handler instanceof AbstractReplyProducingMessageHandler) { + ((AbstractReplyProducingMessageHandler) handler).setRequiresReply(this.requiresReply); + } + else { + if (this.requiresReply && logger.isDebugEnabled()) { + logger.debug("requires-reply can only be set to AbstractReplyProducingMessageHandler " + + "or its subclass, " + handler.getComponentName() + " doesn't support it."); + } + } + } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/FilterFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/FilterFactoryBean.java index 44de23a474..d6d99afeb5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/FilterFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/FilterFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-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. @@ -34,6 +34,8 @@ import org.springframework.util.StringUtils; * @author Mark Fisher * @author Gary Russell * @author David Liu + * @author Artem Bilan + * * @since 2.0 */ public class FilterFactoryBean extends AbstractStandardMessageHandlerFactoryBean { @@ -42,8 +44,6 @@ public class FilterFactoryBean extends AbstractStandardMessageHandlerFactoryBean private volatile Boolean throwExceptionOnRejection; - private volatile Long sendTimeout; - private volatile Boolean discardWithinAdvice; public void setDiscardChannel(MessageChannel discardChannel) { @@ -54,10 +54,6 @@ public class FilterFactoryBean extends AbstractStandardMessageHandlerFactoryBean this.throwExceptionOnRejection = throwExceptionOnRejection; } - public void setSendTimeout(Long sendTimeout) { - this.sendTimeout = sendTimeout; - } - public void setDiscardWithinAdvice(boolean discardWithinAdvice) { this.discardWithinAdvice = discardWithinAdvice; } @@ -113,12 +109,12 @@ public class FilterFactoryBean extends AbstractStandardMessageHandlerFactoryBean @Override protected void postProcessReplyProducer(AbstractMessageProducingHandler handler) { - if (this.sendTimeout != null) { - handler.setSendTimeout(this.sendTimeout); - } + super.postProcessReplyProducer(handler); + if (!(handler instanceof MessageFilter)) { - Assert.isNull(this.throwExceptionOnRejection, "Cannot set throwExceptionOnRejection if the referenced bean is " - + "an AbstractReplyProducingMessageHandler, but not a MessageFilter"); + Assert.isNull(this.throwExceptionOnRejection, + "Cannot set throwExceptionOnRejection if the referenced bean is " + + "an AbstractReplyProducingMessageHandler, but not a MessageFilter"); Assert.isNull(this.discardChannel, "Cannot set discardChannel if the referenced bean is " + "an AbstractReplyProducingMessageHandler, but not a MessageFilter"); Assert.isNull(this.discardWithinAdvice, "Cannot set discardWithinAdvice if the referenced bean is " @@ -137,8 +133,8 @@ public class FilterFactoryBean extends AbstractStandardMessageHandlerFactoryBean protected boolean canBeUsedDirect(AbstractMessageProducingHandler handler) { return handler instanceof MessageFilter || (!(handler instanceof MessageSelector) - && this.discardChannel == null && this.throwExceptionOnRejection == null - && this.discardWithinAdvice == null); + && this.discardChannel == null && this.throwExceptionOnRejection == null + && this.discardWithinAdvice == null); } @Override diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/RouterFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/RouterFactoryBean.java index b7dbc966c7..b90d275382 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/RouterFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/RouterFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-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. @@ -38,6 +38,7 @@ import org.springframework.util.StringUtils; * @author Dave Syer * @author Gary Russell * @author David Liu + * @author Artem Bilan */ public class RouterFactoryBean extends AbstractStandardMessageHandlerFactoryBean { @@ -47,8 +48,6 @@ public class RouterFactoryBean extends AbstractStandardMessageHandlerFactoryBean private volatile String defaultOutputChannelName; - private volatile Long sendTimeout; - private volatile Boolean resolutionRequired; private volatile Boolean applySequence; @@ -63,10 +62,6 @@ public class RouterFactoryBean extends AbstractStandardMessageHandlerFactoryBean this.defaultOutputChannelName = defaultOutputChannelName; } - public void setSendTimeout(Long timeout) { - this.sendTimeout = timeout; - } - public void setResolutionRequired(Boolean resolutionRequired) { this.resolutionRequired = resolutionRequired; } @@ -124,8 +119,8 @@ public class RouterFactoryBean extends AbstractStandardMessageHandlerFactoryBean if (this.defaultOutputChannelName != null) { router.setDefaultOutputChannelName(this.defaultOutputChannelName); } - if (this.sendTimeout != null) { - router.setSendTimeout(this.sendTimeout); + if (getSendTimeout() != null) { + router.setSendTimeout(getSendTimeout()); } if (this.applySequence != null) { router.setApplySequence(this.applySequence); @@ -155,7 +150,7 @@ public class RouterFactoryBean extends AbstractStandardMessageHandlerFactoryBean protected boolean noRouterAttributesProvided() { return this.channelMappings == null && this.defaultOutputChannel == null - && this.sendTimeout == null && this.resolutionRequired == null && this.applySequence == null + && getSendTimeout() == null && this.resolutionRequired == null && this.applySequence == null && this.ignoreSendFailures == null; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ServiceActivatorFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ServiceActivatorFactoryBean.java index ebc0634909..9361320795 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ServiceActivatorFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ServiceActivatorFactoryBean.java @@ -16,9 +16,10 @@ package org.springframework.integration.config; +import java.util.Arrays; + import org.springframework.expression.Expression; import org.springframework.integration.handler.AbstractMessageProducingHandler; -import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor; import org.springframework.integration.handler.MessageProcessor; import org.springframework.integration.handler.ReplyProducingMessageHandlerWrapper; @@ -38,16 +39,10 @@ import org.springframework.util.StringUtils; */ public class ServiceActivatorFactoryBean extends AbstractStandardMessageHandlerFactoryBean { - private volatile Long sendTimeout; + private String[] headers; - private volatile Boolean requiresReply; - - public void setSendTimeout(Long sendTimeout) { - this.sendTimeout = sendTimeout; - } - - public void setRequiresReply(Boolean requiresReply) { - this.requiresReply = requiresReply; + public void setNotPropagatedHeaders(String... headers) { + this.headers = Arrays.copyOf(headers, headers.length); } @Override @@ -119,19 +114,10 @@ public class ServiceActivatorFactoryBean extends AbstractStandardMessageHandlerF @Override protected void postProcessReplyProducer(AbstractMessageProducingHandler handler) { - if (this.sendTimeout != null) { - handler.setSendTimeout(this.sendTimeout); - } - if (this.requiresReply != null) { - if (handler instanceof AbstractReplyProducingMessageHandler) { - ((AbstractReplyProducingMessageHandler) handler).setRequiresReply(this.requiresReply); - } - else { - if (this.requiresReply && logger.isDebugEnabled()) { - logger.debug("requires-reply can only be set to AbstractReplyProducingMessageHandler or its subclass, " - + handler.getComponentName() + " doesn't support it."); - } - } + super.postProcessReplyProducer(handler); + + if (this.headers != null) { + handler.setNotPropagatedHeaders(this.headers); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/SplitterFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/SplitterFactoryBean.java index 5fb5856ec6..79b7de5ef5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/SplitterFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/SplitterFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-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. @@ -18,7 +18,6 @@ package org.springframework.integration.config; import org.springframework.expression.Expression; import org.springframework.integration.handler.AbstractMessageProducingHandler; -import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.splitter.AbstractMessageSplitter; import org.springframework.integration.splitter.DefaultMessageSplitter; import org.springframework.integration.splitter.ExpressionEvaluatingSplitter; @@ -34,30 +33,14 @@ import org.springframework.util.StringUtils; * @author Iwein Fuld * @author Gary Russell * @author David Liu + * @author Artem Bilan */ public class SplitterFactoryBean extends AbstractStandardMessageHandlerFactoryBean { - private volatile Long sendTimeout; - - private volatile Boolean requiresReply; - private volatile Boolean applySequence; private volatile String delimiters; - - public void setSendTimeout(Long sendTimeout) { - this.sendTimeout = sendTimeout; - } - - public boolean isRequiresReply() { - return this.requiresReply; - } - - public void setRequiresReply(boolean requiresReply) { - this.requiresReply = requiresReply; - } - public void setApplySequence(boolean applySequence) { this.applySequence = applySequence; } @@ -116,18 +99,8 @@ public class SplitterFactoryBean extends AbstractStandardMessageHandlerFactoryBe @Override protected void postProcessReplyProducer(AbstractMessageProducingHandler handler) { - if (this.sendTimeout != null) { - handler.setSendTimeout(this.sendTimeout); - } - if (this.requiresReply != null) { - if (handler instanceof AbstractReplyProducingMessageHandler) { - ((AbstractReplyProducingMessageHandler) handler).setRequiresReply(this.requiresReply); - } - else if (this.requiresReply && logger.isDebugEnabled()) { - logger.debug("requires-reply can only be set to AbstractReplyProducingMessageHandler or its subclass, " - + handler.getComponentName() + " doesn't support it."); - } - } + super.postProcessReplyProducer(handler); + if (!(handler instanceof AbstractMessageSplitter)) { Assert.isNull(this.applySequence, "Cannot set applySequence if the referenced bean is " + "an AbstractReplyProducingMessageHandler, but not an AbstractMessageSplitter"); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/TransformerFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/TransformerFactoryBean.java index 075bd1a98a..a4371782a1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/TransformerFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/TransformerFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-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. @@ -32,13 +32,12 @@ import org.springframework.util.StringUtils; * @author Mark Fisher * @author Gary Russell * @author David Liu + * @author Artem Bilan */ public class TransformerFactoryBean extends AbstractStandardMessageHandlerFactoryBean { - private volatile Long sendTimeout; - - public void setSendTimeout(Long sendTimeout) { - this.sendTimeout = sendTimeout; + public TransformerFactoryBean() { + setRequiresReply(true); } @Override @@ -74,13 +73,6 @@ public class TransformerFactoryBean extends AbstractStandardMessageHandlerFactor return handler; } - @Override - protected void postProcessReplyProducer(AbstractMessageProducingHandler handler) { - if (this.sendTimeout != null) { - handler.setSendTimeout(this.sendTimeout); - } - } - /** * Always returns true - any {@link AbstractMessageProducingHandler} can * be used directly. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ServiceActivatorParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ServiceActivatorParser.java index ada8ae3073..dec51c8fc3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ServiceActivatorParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ServiceActivatorParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-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. @@ -28,6 +28,7 @@ import org.springframework.integration.config.ServiceActivatorFactoryBean; * @author Mark Fisher * @author Oleg Zhurakousky * @author Gary Russell + * @author Artme Bilan */ public class ServiceActivatorParser extends AbstractDelegatingConsumerEndpointParser { @@ -44,6 +45,7 @@ public class ServiceActivatorParser extends AbstractDelegatingConsumerEndpointPa @Override void postProcess(BeanDefinitionBuilder builder, Element element, ParserContext parserContext) { IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "async"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "not-propagated-headers"); } } 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 291b007fd1..1496f6c37d 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 @@ -187,7 +187,7 @@ public abstract class ConsumerEndpointSpec, ((AbstractReplyProducingMessageHandler) this.handler).setRequiresReply(requiresReply); } else { - logger.warn("'requiresReply' can be applied only for AbstractReplyProducingMessageHandler"); + this.logger.warn("'requiresReply' can be applied only for AbstractReplyProducingMessageHandler"); } return _this(); } @@ -207,7 +207,7 @@ public abstract class ConsumerEndpointSpec, ((AbstractMessageRouter) this.handler).setSendTimeout(sendTimeout); } else { - logger.warn("'sendTimeout' can be applied only for AbstractMessageProducingHandler"); + this.logger.warn("'sendTimeout' can be applied only for AbstractMessageProducingHandler"); } return _this(); } @@ -223,7 +223,7 @@ public abstract class ConsumerEndpointSpec, ((AbstractMessageHandler) this.handler).setOrder(order); } else { - logger.warn("'order' can be applied only for AbstractMessageHandler"); + this.logger.warn("'order' can be applied only for AbstractMessageHandler"); } return _this(); } @@ -244,7 +244,26 @@ public abstract class ConsumerEndpointSpec, ((AbstractMessageProducingHandler) this.handler).setAsync(async); } else { - logger.warn("'async' can be applied only for AbstractMessageProducingHandler"); + this.logger.warn("'async' can be applied only for AbstractMessageProducingHandler"); + } + return _this(); + } + + /** + * Set header patterns ("xxx*", "*xxx", "*xxx*" or "xxx*yyy") + * that will NOT be copied from the inbound message. + * At least one pattern as "*" means do not copy headers at all. + * @param headerPatterns the headers to not propagate from the inbound message. + * @return the endpoint spec. + * @see AbstractMessageProducingHandler#setNotPropagatedHeaders(String...) + */ + public S notPropagatedHeaders(String... headerPatterns) { + assertHandler(); + if (this.handler instanceof AbstractMessageProducingHandler) { + ((AbstractMessageProducingHandler) this.handler).setNotPropagatedHeaders(headerPatterns); + } + else { + this.logger.warn("'headerPatterns' can be applied only for AbstractMessageProducingHandler"); } return _this(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java index fc95b9f01f..04e0fa26f3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java @@ -43,6 +43,7 @@ import org.springframework.messaging.core.DestinationResolutionException; import org.springframework.messaging.support.ErrorMessage; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; +import org.springframework.util.PatternMatchUtils; import org.springframework.util.StringUtils; import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.ListenableFutureCallback; @@ -64,18 +65,20 @@ import reactor.core.publisher.Mono; public abstract class AbstractMessageProducingHandler extends AbstractMessageHandler implements MessageProducer, HeaderPropagationAware { - private final Set notPropagatedHeaders = new HashSet(); - protected final MessagingTemplate messagingTemplate = new MessagingTemplate(); - private volatile MessageChannel outputChannel; + private boolean async; - private volatile String outputChannelName; + private String outputChannelName; - private volatile boolean async; + private MessageChannel outputChannel; + + private String[] notPropagatedHeaders; private boolean selectiveHeaderPropagation; + private boolean noHeadersPropagation; + /** * Set the timeout for sending reply Messages. * @param sendTimeout The send timeout. @@ -115,10 +118,13 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan } /** - * Set headers that will NOT be copied from the inbound message if + * Set header patterns ("xxx*", "*xxx", "*xxx*" or "xxx*yyy") + * that will NOT be copied from the inbound message if * {@link #shouldCopyRequestHeaders() shouldCopyRequestHeaaders} is true. + * At least one pattern as "*" means do not copy headers at all. * @param headers the headers to not propagate from the inbound message. * @since 4.3.10 + * @see org.springframework.util.PatternMatchUtils */ @Override public void setNotPropagatedHeaders(String... headers) { @@ -126,30 +132,48 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan } private void updateNotPropagatedHeaders(String[] headers, boolean merge) { + Set headerPatterns = new HashSet<>(); + + if (merge) { + headerPatterns.addAll(Arrays.asList(this.notPropagatedHeaders)); + } + if (!ObjectUtils.isEmpty(headers)) { Assert.noNullElements(headers, "null elements are not allowed in 'headers'"); - if (!merge) { - this.notPropagatedHeaders.clear(); - } - this.notPropagatedHeaders.addAll(Arrays.asList(headers)); + + headerPatterns.addAll(Arrays.asList(headers)); + + this.notPropagatedHeaders = headerPatterns.toArray(new String[headerPatterns.size()]); } - this.selectiveHeaderPropagation = this.notPropagatedHeaders.size() > 0; + + boolean hasAsterisk = headerPatterns.contains("*"); + + if (hasAsterisk) { + this.notPropagatedHeaders = new String[] { "*" }; + this.noHeadersPropagation = true; + } + + this.selectiveHeaderPropagation = this.notPropagatedHeaders.length > 0; } /** - * Get the header names this handler doesn't propagate. + * Get the header patterns this handler doesn't propagate. * @return an immutable {@link java.util.Collection} of headers that will not be * copied from the inbound message if {@link #shouldCopyRequestHeaders()} is true. * @since 4.3.10 * @see #setNotPropagatedHeaders(String...) + * @see org.springframework.util.PatternMatchUtils */ @Override public Collection getNotPropagatedHeaders() { - return Collections.unmodifiableSet(this.notPropagatedHeaders); + return this.notPropagatedHeaders != null + ? Collections.unmodifiableSet(new HashSet<>(Arrays.asList(this.notPropagatedHeaders))) + : Collections.emptyList(); } /** - * Add headers that will NOT be copied from the inbound message if + * Add header patterns ("xxx*", "*xxx", "*xxx*" or "xxx*yyy") + * that will NOT be copied from the inbound message if * {@link #shouldCopyRequestHeaders()} is true, instead of overwriting the existing * set. * @param headers the headers to not propagate from the inbound message. @@ -344,7 +368,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan protected Message createOutputMessage(Object output, MessageHeaders requestHeaders) { AbstractIntegrationMessageBuilder builder = null; if (output instanceof Message) { - if (!this.shouldCopyRequestHeaders()) { + if (this.noHeadersPropagation || !shouldCopyRequestHeaders()) { return (Message) output; } builder = this.getMessageBuilderFactory().fromMessage((Message) output); @@ -355,12 +379,13 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan else { builder = this.getMessageBuilderFactory().withPayload(output); } - if (this.shouldCopyRequestHeaders()) { + if (!this.noHeadersPropagation && shouldCopyRequestHeaders()) { if (this.selectiveHeaderPropagation) { - Map headersToCopy = new HashMap(requestHeaders); - for (String header : this.notPropagatedHeaders) { - headersToCopy.remove(header); - } + Map headersToCopy = new HashMap<>(requestHeaders); + + headersToCopy.entrySet() + .removeIf(entry -> PatternMatchUtils.simpleMatch(this.notPropagatedHeaders, entry.getKey())); + builder.copyHeadersIfAbsent(headersToCopy); } else { diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd index bf082ed31e..3d02d23d59 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd @@ -1212,16 +1212,7 @@ - - - - Specify whether the service method must return a non-null value. This value will be - 'false' by default, but if set to 'true', a ReplyRequiredException will be thrown when - the underlying service method (or expression) returns a null value. - - - - + and this flag is 'true', the calling @@ -1231,6 +1222,15 @@ ]]> + + + + Header patterns ("xxx*", "*xxx", "*xxx*" or "xxx*yyy") + that will NOT be copied from the inbound message. + '*' means do not copy headers at all. + + + @@ -3617,15 +3617,6 @@ - - - - Specify whether the service method must return a non-null value. This value will be - 'false' by default, but if set to 'true', a ReplyRequiredException will be thrown when - the underlying service method (or expression) returns a null value. - - - @@ -4190,6 +4181,20 @@ + + + + If set to 'true', a reply must return a non-null value. + By setting 'requires-reply' to 'true', a 'ReplyRequiredException' + will be raised for null reply messages. If 'requires-reply' is set + to false, those messages are silently dropped. + This attribute defaults to 'true' for 'transformer'. + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandlerTests.java index e9c342579f..3c253c0346 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandlerTests.java @@ -46,6 +46,7 @@ import org.springframework.messaging.support.GenericMessage; * @author Gunnar Hillert * @author Gary Russell * @author Marius Bogoevici + * @author Artem Bilan */ @RunWith(MockitoJUnitRunner.class) public class AbstractReplyProducingMessageHandlerTests { @@ -80,19 +81,20 @@ public class AbstractReplyProducingMessageHandlerTests { } @Test + @SuppressWarnings("unchecked") public void testNotPropagate() { AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() { @Override protected Object handleRequestMessage(Message requestMessage) { - return new GenericMessage("world", Collections.singletonMap("bar", "RAB")); + return new GenericMessage<>("world", Collections.singletonMap("bar", "RAB")); } }; assertThat(handler.getNotPropagatedHeaders(), emptyCollectionOf(String.class)); - handler.setNotPropagatedHeaders("foo", "bar"); + handler.setNotPropagatedHeaders("f*", "*r"); handler.setOutputChannel(this.channel); - assertThat(handler.getNotPropagatedHeaders(), containsInAnyOrder("foo", "bar")); + assertThat(handler.getNotPropagatedHeaders(), containsInAnyOrder("f*", "*r")); ArgumentCaptor> captor = ArgumentCaptor.forClass(Message.class); willReturn(true).given(this.channel).send(captor.capture()); handler.handleMessage(MessageBuilder.withPayload("hello") @@ -120,9 +122,9 @@ public class AbstractReplyProducingMessageHandlerTests { }; assertThat(handler.getNotPropagatedHeaders(), emptyCollectionOf(String.class)); handler.setNotPropagatedHeaders("foo"); - handler.addNotPropagatedHeaders("bar"); + handler.addNotPropagatedHeaders("b*r"); handler.setOutputChannel(this.channel); - assertThat(handler.getNotPropagatedHeaders(), containsInAnyOrder("foo", "bar")); + assertThat(handler.getNotPropagatedHeaders(), containsInAnyOrder("foo", "b*r")); ArgumentCaptor> captor = (ArgumentCaptor>) (ArgumentCaptor) ArgumentCaptor.forClass(Message.class); willReturn(true).given(this.channel).send(captor.capture()); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-context.xml index 961b951711..4005c18515 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-context.xml @@ -38,7 +38,10 @@ class="org.springframework.integration.handler.ServiceActivatorDefaultFrameworkMethodTests$TestMessageHandler"/> - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java index f911c30914..3269509dd0 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java @@ -19,6 +19,7 @@ package org.springframework.integration.handler; import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.instanceOf; +import static org.hamcrest.Matchers.not; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; @@ -27,6 +28,7 @@ import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import static org.springframework.integration.test.matcher.HeaderMatcher.hasHeaderKey; import java.util.concurrent.atomic.AtomicReference; @@ -189,18 +191,20 @@ public class ServiceActivatorDefaultFrameworkMethodTests { this.handlerTestInputChannel.send(message); } - // INT-2399 @Test public void testMessageProcessor() { Object processor = TestUtils.getPropertyValue(processorTestService, "handler.processor"); assertSame(testMessageProcessor, processor); QueueChannel replyChannel = new QueueChannel(); - Message message = MessageBuilder.withPayload("bar").setReplyChannel(replyChannel).build(); + Message message = MessageBuilder.withPayload("bar") + .setReplyChannel(replyChannel) + .setHeader("foo", "foo") + .build(); this.processorTestInputChannel.send(message); Message reply = replyChannel.receive(0); assertEquals("foo:bar", reply.getPayload()); - assertEquals("processorTestInputChannel,processorTestService", reply.getHeaders().get("history").toString()); + assertThat(reply, not(hasHeaderKey("foo"))); } @Test diff --git a/spring-integration-core/src/test/java/org/springframework/integration/transformer/TransformerContextTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/transformer/TransformerContextTests-context.xml index 169b77bd85..4691c6bcc0 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/transformer/TransformerContextTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/transformer/TransformerContextTests-context.xml @@ -25,7 +25,8 @@ - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/transformer/TransformerContextTests.java b/spring-integration-core/src/test/java/org/springframework/integration/transformer/TransformerContextTests.java index 0bd80f978e..af93832242 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/transformer/TransformerContextTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/transformer/TransformerContextTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-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. @@ -18,6 +18,7 @@ package org.springframework.integration.transformer; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import org.junit.Test; @@ -90,6 +91,9 @@ public class TransformerContextTests { assertFalse(this.testBean.isRunning()); this.pojoTransformer.start(); assertTrue(this.testBean.isRunning()); + + this.directRef.send(new GenericMessage("bar")); + assertNull(this.output.receive(0)); } public static class FooAdvice extends AbstractRequestHandlerAdvice { @@ -106,6 +110,9 @@ public class TransformerContextTests { @Override protected Object handleRequestMessage(Message requestMessage) { + if ("bar".equals(requestMessage.getPayload())) { + return null; + } Exception e = new RuntimeException(); StackTraceElement[] st = e.getStackTrace(); return MessageBuilder.withPayload(requestMessage.getPayload().toString().toUpperCase()) diff --git a/src/reference/asciidoc/message.adoc b/src/reference/asciidoc/message.adoc index 421ffd9ccf..670a88f5fe 100644 --- a/src/reference/asciidoc/message.adoc +++ b/src/reference/asciidoc/message.adoc @@ -394,9 +394,16 @@ Also, a header is only propagated if it does not already exist in the outbound m Starting with _version 4.3.10_, you can configure message handlers (that modify messages and produce output) to suppress the propagation of specific headers. Call the `setNotPropagatedHeaders()` or `addNotPropagatedHeaders()` methods on the `MessageProducingMessageHandler` abstract class, to configure the header(s) you don't want to be copied. + You can also globally suppress propagation of specific message headers by setting the `readOnlyHeaders` property in `META-INF/spring.integration.properties` to a comma-delimited list of headers. -IMPORTANT: Header propagation suppression does not apply to those endpoints that don't modify the message, e.g. <> and <> +Starting with _version 5.0_, the `setNotPropagatedHeaders()` implementation on the `AbstractMessageProducingHandler` applies the simple patterns (`xxx*`, `*xxx`, `*xxx*` or `xxx*yyy`) to allow filtering headers with a common suffix or prefix. +See `PatternMatchUtils` JavaDocs for more information. +When one of the patterns is `*` (asterisk), no headers are propagated; all other patterns are ignored. +In this case the Service Activator behaves the same way as Transformer and any required headers must be supplied in the `Message` returned from the service method. +The option `notPropagatedHeaders()` is available in the `ConsumerEndpointSpec` for Java DSL, as well as for XML configuration of the `` component as a `not-propagated-headers` attribute. + +IMPORTANT: Header propagation suppression does not apply to those endpoints that don't modify the message, e.g. <> and <>. [[message-implementations]]