INT-4365: Improve notPropagatedHeaders function

JIRA: https://jira.spring.io/browse/INT-4365

It is much useful to configure the `notPropagatedHeaders` as a set of
patterns to match.
In this case we can filter a group of headers with the common prefix or
suffix

* Allow to configure `AbstractMessageProducingHandler.setNotPropagatedHeaders`
as simple patterns; the `*` means filter all - not copy request headers
at all - similar to `transformer` behavior
* Add `ConsumerEndpointSpec.notPropagatedHeaders()` for Java DSL
* Add `not-propagated-headers` to the `<service-activator>`

Address PR comments; some other improvements

* Fix `ConsumerEndpointSpec#notPropagatedHeaders()` log message
* Improve `AbstractMessageProducingHandler.notPropagatedHeaders() logic
so any `*` in the set of patterns eliminates all others since it has
a highest priority
* Expose `requires-reply` for the `<transformer>` as a `true` by default
* Refactor a bit `AbstractStandardMessageHandlerFactoryBean` hierarchy to
avoid duplicated code

* Fix `message.adoc`
* Add `noHeadersPropagation` flag to the `AbstractMessageProducingHandler`
* Rework logic in the `updateNotPropagatedHeaders()` to store the array
of patterns instead of `Set` to avoid extra operation on each message
* Combine `noHeadersPropagation` with the `shouldCopyRequestHeaders()`
in the `createOutputMessage()` for logic to determine if we should start
the copy-headers procedure at all

* Revert `AbstractMessageProducingHandler.selectiveHeaderPropagation`
* Optimize `AmqpOutboundGatewayParserTests` performance from 3 secs to 0.5

Fix more NPEs in the `AbstractMessageProducingHandler`
This commit is contained in:
Artem Bilan
2017-11-17 16:53:19 -05:00
committed by Gary Russell
parent 7c701ca5e6
commit 90c46f5a79
17 changed files with 230 additions and 199 deletions

View File

@@ -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();
}

View File

@@ -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<MessageHandler> {
private static final ExpressionParser expressionParser = new SpelExpressionParser(new SpelParserConfiguration(true,
true));
private static final ExpressionParser expressionParser = new SpelExpressionParser();
private static final Set<MessageHandler> referencedReplyProducers = new HashSet<MessageHandler>();
private static final Set<MessageHandler> 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.");
}
}
}
}
}

View File

@@ -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

View File

@@ -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;
}

View File

@@ -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);
}
}

View File

@@ -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");

View File

@@ -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.

View File

@@ -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");
}
}

View File

@@ -187,7 +187,7 @@ public abstract class ConsumerEndpointSpec<S extends ConsumerEndpointSpec<S, H>,
((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<S extends ConsumerEndpointSpec<S, H>,
((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<S extends ConsumerEndpointSpec<S, H>,
((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<S extends ConsumerEndpointSpec<S, H>,
((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();
}

View File

@@ -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<String> notPropagatedHeaders = new HashSet<String>();
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<String> 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<String> 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<String, Object> headersToCopy = new HashMap<String, Object>(requestHeaders);
for (String header : this.notPropagatedHeaders) {
headersToCopy.remove(header);
}
Map<String, Object> headersToCopy = new HashMap<>(requestHeaders);
headersToCopy.entrySet()
.removeIf(entry -> PatternMatchUtils.simpleMatch(this.notPropagatedHeaders, entry.getKey()));
builder.copyHeadersIfAbsent(headersToCopy);
}
else {

View File

@@ -1212,16 +1212,7 @@
<xsd:complexType name="serviceActivatorType">
<xsd:complexContent>
<xsd:extension base="expressionOrInnerEndpointDefinitionAware">
<xsd:attribute name="requires-reply" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="async" type="xsd:string" use="optional">
<xsd:attribute name="async">
<xsd:annotation>
<xsd:documentation><![CDATA[
If the service method returns a ListenableFuture<?> and this flag is 'true', the calling
@@ -1231,6 +1222,15 @@
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="not-propagated-headers">
<xsd:annotation>
<xsd:documentation>
Header patterns ("xxx*", "*xxx", "*xxx*" or "xxx*yyy")
that will NOT be copied from the inbound message.
'*' means do not copy headers at all.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
@@ -3617,15 +3617,6 @@
<xsd:complexType name="splitter-type">
<xsd:complexContent>
<xsd:extension base="expressionOrInnerEndpointDefinitionAware">
<xsd:attribute name="requires-reply" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="apply-sequence" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation>
@@ -4190,6 +4181,20 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="requires-reply">
<xsd:annotation>
<xsd:documentation>
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'.
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string" />
</xsd:simpleType>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>

View File

@@ -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<String>("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<Message<?>> 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<Message<?>> captor =
(ArgumentCaptor<Message<?>>) (ArgumentCaptor<?>) ArgumentCaptor.forClass(Message.class);
willReturn(true).given(this.channel).send(captor.capture());

View File

@@ -38,7 +38,10 @@
class="org.springframework.integration.handler.ServiceActivatorDefaultFrameworkMethodTests$TestMessageHandler"/>
</service-activator>
<service-activator id="processorTestService" input-channel="processorTestInputChannel" ref="testMessageProcessor"/>
<service-activator id="processorTestService"
input-channel="processorTestInputChannel"
ref="testMessageProcessor"
not-propagated-headers="*"/>
<gateway id="gateway" default-request-channel="requestChannel" default-reply-channel="replyChannel"/>

View File

@@ -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

View File

@@ -25,7 +25,8 @@
<beans:bean class="org.springframework.integration.transformer.TransformerContextTests$Bar"/>
</transformer>
<transformer input-channel="directRef" output-channel="output" ref="trans" method="handleMessage"/>
<transformer input-channel="directRef" output-channel="output" ref="trans" method="handleMessage"
requires-reply="false"/>
<beans:bean id="trans" class="org.springframework.integration.transformer.TransformerContextTests$Bar"/>

View File

@@ -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<String>("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())

View File

@@ -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. <<bridge, bridges>> and <<router, routers>>
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 `<service-activator>` 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. <<bridge, bridges>> and <<router, routers>>.
[[message-implementations]]