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

INT-2243: Delayer: Add 'expression' Support

Browse Source 
Previously, the `<delayer>` provided a `delay-header-name` attribute.
In complex cases there was need to precede it with `<header-enricher>`.

* Add support for an 'expression' attribute and sub-element
* Deprecate `delay-header-name`
* Make `DelayHandler.DelayedMessageWrapper` *public* to allow access for Messages in the Store
* Add tests
* Add 'What's new' section
* Polishing Delayer's doc regarding new abilities

JIRA: https://jira.springsource.org/browse/INT-2243, https://jira.springsource.org/browse/INT-3049

INT-2243: ban delay-header-name with expression

INT-2243 Polishing

INT-2243: fall-back to default on Eval Exception

INT-2243: DelayedMessageWrapper refactoring

* Make `DelayedMessageWrapper` Spring Data Mongo mapping compatible.
In terms of Spring Data - add Persistence Constructor

INT-2243 add 'ignore-expression-failures' support

* Add `ignore-expression-failures` to the `<delayer>`
* Add tests for `ignore-expression-failures`
* Add a note to the RM
* Describe SpEL side-effects for `DelayHandler`

Doc Polishing
This commit is contained in:
Artem Bilan
2013-06-15 17:51:58 +03:00
committed by Gary Russell
parent 67fd4a5a60
commit afe56ca6d5
12 changed files with 329 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
spring-integration-core/src/main/java/org/springframework/integration/config/xml/DelayerParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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,8 @@ package org.springframework.integration.config.xml;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.ExpressionFactoryBean;
import org.springframework.integration.expression.DynamicExpression;
import org.springframework.integration.handler.DelayHandler;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
@@ -43,13 +45,22 @@ public class DelayerParser extends AbstractConsumerEndpointParser {
String defaultDelay = element.getAttribute("default-delay");
String delayHeaderName = element.getAttribute("delay-header-name");
String expression = element.getAttribute(EXPRESSION_ATTRIBUTE);
Element expressionElement = DomUtils.getChildElementByTagName(element, "expression");
boolean hasDefaultDelay = StringUtils.hasText(defaultDelay);
boolean hasDelayHeaderName = StringUtils.hasText(delayHeaderName);
boolean hasExpression = StringUtils.hasText(expression);
boolean hasExpressionElement = expressionElement != null;
if (!(hasDefaultDelay | hasDelayHeaderName)) {
if (!(hasDefaultDelay | hasDelayHeaderName | hasExpression | hasExpressionElement)) {
parserContext.getReaderContext()
.error("The 'default-delay' or 'delay-header-name' attributes should be provided.", element);
.error("The 'default-delay' or 'delay-header-name', or 'expression' attributes, or 'expression' sub-element should be provided.", element);
}
if ((hasDelayHeaderName & (hasExpression | hasExpressionElement)) | (hasExpression & hasExpressionElement)) {
parserContext.getReaderContext()
.error("'delay-header-name', 'expression' attribute and 'expression' sub-element are mutually exclusive.", element);
}
builder.addConstructorArgValue(id + ".messageGroupId");
@@ -62,12 +73,32 @@ public class DelayerParser extends AbstractConsumerEndpointParser {
if (hasDefaultDelay) {
builder.addPropertyValue("defaultDelay", defaultDelay);
}
BeanDefinitionBuilder expressionBuilder = null;
if (hasExpression) {
expressionBuilder = BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class);
expressionBuilder.addConstructorArgValue(expression);
}
else if(expressionElement != null) {
expressionBuilder = BeanDefinitionBuilder.genericBeanDefinition(
DynamicExpression.class);
String key = expressionElement.getAttribute("key");
String expressionSourceReference = expressionElement.getAttribute("source");
expressionBuilder.addConstructorArgValue(key);
expressionBuilder.addConstructorArgReference(expressionSourceReference);
}
if (expressionBuilder != null) {
builder.addPropertyValue("delayExpression", expressionBuilder.getBeanDefinition());
}
if (hasDelayHeaderName) {
builder.addPropertyValue("delayHeaderName", delayHeaderName);
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-store");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "ignore-expression-failures");
Element txElement = DomUtils.getChildElementByTagName(element, "transactional");
Element adviceChainElement = DomUtils.getChildElementByTagName(element, "advice-chain");

127
spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java
View File

@@ -22,13 +22,22 @@ import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import org.aopalliance.aop.Advice;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.EvaluationException;
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.Message;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.MessagingException;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.MessageStore;
@@ -43,7 +52,7 @@ import org.springframework.util.CollectionUtils;
/**
* A {@link MessageHandler} that is capable of delaying the continuation of a
* Message flow based on the presence of a delay header on an inbound Message
* Message flow based on the result of evaluation {@code delayExpression} on an inbound {@link Message}
* or a default delay value configured on this handler. Note that the
* continuation of the flow is delegated to a {@link TaskScheduler}, and
* therefore, the calling thread does not block. The advantage of this approach
@@ -55,15 +64,15 @@ import org.springframework.util.CollectionUtils;
* is a side-effect of passing the Message to the output channel after the
* delay with a different Thread in control.
* <p>
* When this handler's 'delayHeaderName' property is configured, that value, if
* present on a Message, will take precedence over the handler's 'defaultDelay'
* value. The actual header value may be a long, a String that can be parsed
* When this handler's {@code delayExpression} property is configured, that evaluation result value
* will take precedence over the handler's {@code defaultDelay} value.
* The actual evaluation result value may be a long, a String that can be parsed
* as a long, or a Date. If it is a long, it will be interpreted as the length
* of time to delay in milliseconds counting from the current time (e.g. a
* value of 5000 indicates that the Message can be released as soon as five
* seconds from the current time). If the value is a Date, it will be
* delayed at least until that Date occurs (i.e. the delay in that case is
* equivalent to <code>headerDate.getTime() - new Date().getTime()</code>).
* equivalent to {@code headerDate.getTime() - new Date().getTime()}).
*
* @author Mark Fisher
* @author Artem Bilan
@@ -71,12 +80,19 @@ import org.springframework.util.CollectionUtils;
*/
@ManagedResource
public class DelayHandler extends AbstractReplyProducingMessageHandler implements DelayHandlerManagement, ApplicationListener<ContextRefreshedEvent> {
public class DelayHandler extends AbstractReplyProducingMessageHandler implements DelayHandlerManagement,
ApplicationListener<ContextRefreshedEvent> {
private static final ExpressionParser expressionParser = new SpelExpressionParser(new SpelParserConfiguration(true, true));
private final String messageGroupId;
private volatile long defaultDelay;
private Expression delayExpression;
private volatile boolean ignoreExpressionFailures = true;
private volatile String delayHeaderName;
private volatile MessageGroupStore messageStore;
@@ -87,6 +103,8 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
private volatile MessageHandler releaseHandler = new ReleaseMessageHandler();
private EvaluationContext evaluationContext;
/**
* Create a DelayHandler with the given 'messageGroupId' that is used as 'key' for {@link MessageGroup}
* to store delayed Messages in the {@link MessageGroupStore}. The sending of Messages after
@@ -109,10 +127,10 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
}
/**
* Set the default delay in milliseconds. If no 'delayHeaderName' property
* Set the default delay in milliseconds. If no {@code delayExpression} property
* has been provided, the default delay will be applied to all Messages. If
* a delay should <em>only</em> be applied to Messages with a
* header, then set this value to 0.
* a delay should <em>only</em> be applied to Messages with evaluation result from
* @code delayExpression}, then set this value to 0.
*/
public void setDefaultDelay(long defaultDelay) {
this.defaultDelay = defaultDelay;
@@ -122,11 +140,36 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
* Specify the name of the header that should be checked for a delay period
* (in milliseconds) or a Date to delay until. If this property is set, any
* such header value will take precedence over this handler's default delay.
* @deprecated in favor of {@link #delayExpression}
*/
@Deprecated
public void setDelayHeaderName(String delayHeaderName) {
this.delayHeaderName = delayHeaderName;
}
/**
* Specify the {@link Expression} that should be checked for a delay period
* (in milliseconds) or a Date to delay until. If this property is set, the
* result of the expression evaluation will take precedence over this handler's default delay.
*/
public void setDelayExpression(Expression delayExpression) {
this.delayExpression = delayExpression;
}
/**
* Specify whether {@code Exceptions} thrown by {@link #delayExpression} evaluation should be
* ignored (only logged). In this case case the delayer will fall back to the
* to the {@link #defaultDelay}.
* If this property is specified as {@code false}, any {@link #delayExpression} evaluation
* {@code Exception} will be thrown to the caller without falling back to the to the {@link #defaultDelay}.
* Default is {@code true}.
*
* @see #determineDelayForMessage
*/
public void setIgnoreExpressionFailures(boolean ignoreExpressionFailures) {
this.ignoreExpressionFailures = ignoreExpressionFailures;
}
/**
* Specify the {@link MessageGroupStore} that should be used to store Messages
* while awaiting the delay.
@@ -161,7 +204,18 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
else {
Assert.isInstanceOf(MessageStore.class, this.messageStore);
}
if (this.delayHeaderName != null) {
logger.warn("'delayHeaderName' is deprecated in favor of 'delayExpression'");
if (this.delayExpression == null) {
this.delayExpression = expressionParser.parseExpression("headers['" + this.delayHeaderName + "']");
}
}
if (this.getBeanFactory() != null) {
this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(this.getBeanFactory());
}
else {
this.evaluationContext = ExpressionUtils.createStandardEvaluationContext();
}
this.releaseHandler = this.createReleaseMessageTask();
}
@@ -185,7 +239,7 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
* and if {@code delay > 0} schedules 'releaseMessage' task after 'delay'.
*
* @param requestMessage - the Message which may be delayed.
* @return - <code>null</code> if 'requestMessage' is delayed,
* @return - {@code null} if 'requestMessage' is delayed,
* otherwise - 'payload' from 'requestMessage'.
*
* @see #releaseMessage
@@ -210,21 +264,37 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
private long determineDelayForMessage(Message<?> message) {
long delay = this.defaultDelay;
if (this.delayHeaderName != null) {
Object headerValue = message.getHeaders().get(this.delayHeaderName);
if (headerValue instanceof Date) {
delay = ((Date) headerValue).getTime() - new Date().getTime();
if (this.delayExpression != null) {
Exception delayValueException = null;
Object delayValue = null;
try {
delayValue = this.delayExpression.getValue(this.evaluationContext, message);
}
else if (headerValue != null) {
catch (EvaluationException e) {
delayValueException = e;
}
if (delayValue instanceof Date) {
delay = ((Date) delayValue).getTime() - new Date().getTime();
}
else if (delayValue != null) {
try {
delay = Long.valueOf(headerValue.toString());
delay = Long.valueOf(delayValue.toString());
}
catch (NumberFormatException e) {
delayValueException = e;
}
}
if (delayValueException != null) {
if (this.ignoreExpressionFailures) {
if (logger.isDebugEnabled()) {
logger.debug("Failed to parse delay from header value '" + headerValue.toString() +
"', will fall back to default delay: " + this.defaultDelay);
logger.debug("Failed to get delay value from 'delayExpression': " + delayValueException.getMessage() +
". Will fall back to default delay: " + this.defaultDelay);
}
}
else {
throw new MessageHandlingException(message, "Error occurred during 'delay' value determination", delayValueException);
}
}
}
return delay;
@@ -238,7 +308,7 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
messageWrapper = (DelayedMessageWrapper) message.getPayload();
}
else {
messageWrapper = new DelayedMessageWrapper(message);
messageWrapper = new DelayedMessageWrapper(message, System.currentTimeMillis());
delayedMessage = MessageBuilder.withPayload(messageWrapper).copyHeaders(message.getHeaders()).build();
this.messageStore.addMessageToGroup(this.messageGroupId, delayedMessage);
}
@@ -277,7 +347,7 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
/**
* Used for reading persisted Messages in the 'messageStore'
* to reschedule them e.g. upon application restart.
* The logic is based on iteration over 'messageGroup.getMessages()'
* The logic is based on iteration over {@code messageGroup.getMessages()}
* and schedules task about 'delay' logic.
* This behavior is dictated by the avoidance of invocation thread overload.
*/
@@ -333,16 +403,17 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
}
private static final class DelayedMessageWrapper implements Serializable {
public static final class DelayedMessageWrapper implements Serializable {
private static final long serialVersionUID = -4739802369074947045L;
private final long requestDate = System.currentTimeMillis();
private final long requestDate;
private final Message<?> original;
public DelayedMessageWrapper(Message<?> original) {
DelayedMessageWrapper(Message<?> original, long requestDate) {
this.original = original;
this.requestDate = requestDate;
}
public long getRequestDate() {
@@ -355,8 +426,12 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
DelayedMessageWrapper that = (DelayedMessageWrapper) o;

40
spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd
View File

@@ -1359,8 +1359,8 @@
<xsd:documentation>
Defines an endpoint that passes a Message to the output-channel after a
delay. The delay may
be
retrieved from a Message header or else fallback to the
be dynamically determined by evaluating an expression
(such as a Message header) or fallback to the
'default-delay' of this endpoint.
</xsd:documentation>
</xsd:annotation>
@@ -1384,6 +1384,14 @@
to proxy DelayHandler's 'release Message task'.
</xsd:documentation>
</xsd:annotation>
<xsd:element name="expression" type="innerExpressionType" minOccurs="0" maxOccurs="1" >
<xsd:annotation>
<xsd:documentation>
Specify the SpEL expression that evaluates to the delay value in milliseconds,
or a java.util.Date.
</xsd:documentation>
</xsd:annotation>
</xsd:element>
<xsd:element name="transactional" type="transactionalType" minOccurs="0" maxOccurs="1" />
<xsd:element name="advice-chain" type="adviceChainType" minOccurs="0" maxOccurs="1" />
</xsd:choice>
@@ -1393,22 +1401,40 @@
Specify the default delay in milliseconds. This value can be set to 0
if the only Messages
that
should be delayed are those with a particular header (in that
case, be sure to provide
a value for the
'delay-header-name' attribute).
should be delayed are those with a particular expression evaluation result.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Specify the SpEL expression that evaluates to the delay value in milliseconds,
or a java.util.Date.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="ignore-expression-failures" type="xsd:string" default="true">
<xsd:annotation>
<xsd:documentation>
Specify whether Exceptions thrown by 'expression' evaluation should be
ignored (only logged). In this case case the delayer will fall back to the
to the 'default-delay'. Default behaviour.
If this attribute is specified as 'false', any
'expression' evaluation Exception will be thrown to the caller without
falling back to the to the 'default-delay'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="delay-header-name" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Specify the name of the header that should contain the delay value.
[DEPRECATED]Specify the name of the header that should contain the delay value.
This value can either
represent the number of milliseconds to delay counting from the current
time or it can be an
absolute Date until
which the Message should be delayed.
This attribute is deprecated in favor of an 'expression' attribute or sub-element.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>

16
spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerParserTests-context.xml
View File

@@ -57,6 +57,22 @@
</advice-chain>
</delayer>
<delayer id="delayerWithExpression"
input-channel="input"
output-channel="output"
expression="100"
ignore-expression-failures="false"/>
<beans:bean id="delayerSource"
class="org.springframework.integration.expression.ReloadableResourceBundleExpressionSource"
p:basename="org/springframework/integration/config/xml/delayer-expression"/>
<delayer id="delayerWithExpressionSubElement"
input-channel="input"
output-channel="output">
<expression key="delay" source="delayerSource"/>
</delayer>
<beans:bean id="testScheduler" class="org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler"
p:poolSize="7"
p:waitForTasksToCompleteOnShutdown="true"/>

37
spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerParserTests.java
View File

@@ -32,6 +32,7 @@ import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.core.Ordered;
import org.springframework.expression.Expression;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.handler.DelayHandler;
import org.springframework.integration.test.util.TestUtils;
@@ -58,19 +59,16 @@ public class DelayerParserTests {
@Test
public void defaultScheduler() {
Object endpoint = context.getBean("delayerWithDefaultScheduler");
assertEquals(EventDrivenConsumer.class, endpoint.getClass());
Object handler = TestUtils.getPropertyValue(endpoint, "handler");
assertEquals(DelayHandler.class, handler.getClass());
DelayHandler delayHandler = (DelayHandler) handler;
DelayHandler delayHandler = context.getBean("delayerWithDefaultScheduler.handler", DelayHandler.class);
assertEquals(99, delayHandler.getOrder());
DirectFieldAccessor accessor = new DirectFieldAccessor(delayHandler);
assertEquals(context.getBean("output"), accessor.getPropertyValue("outputChannel"));
assertEquals(new Long(1234), accessor.getPropertyValue("defaultDelay"));
assertEquals("foo", accessor.getPropertyValue("delayHeaderName"));
assertEquals(new Long(987), new DirectFieldAccessor(
accessor.getPropertyValue("messagingTemplate")).getPropertyValue("sendTimeout"));
assertNull(accessor.getPropertyValue("taskScheduler"));
assertEquals(context.getBean("output"), TestUtils.getPropertyValue(delayHandler, "outputChannel"));
assertEquals(new Long(1234), TestUtils.getPropertyValue(delayHandler, "defaultDelay", Long.class));
assertEquals("foo", TestUtils.getPropertyValue(delayHandler, "delayHeaderName"));
//INT-2243
assertNotNull(TestUtils.getPropertyValue(delayHandler, "delayExpression"));
assertEquals("headers['foo']", TestUtils.getPropertyValue(delayHandler, "delayExpression", Expression.class).getExpressionString());
assertEquals(new Long(987), TestUtils.getPropertyValue(delayHandler, "messagingTemplate.sendTimeout", Long.class));
assertNull(TestUtils.getPropertyValue(delayHandler, "taskScheduler"));
}
@Test
@@ -134,4 +132,19 @@ public class DelayerParserTests {
assertEquals("{*=PROPAGATION_REQUIRES_NEW,ISOLATION_DEFAULT,readOnly}", nameMap.toString());
}
@Test
public void testInt2243Expression() {
DelayHandler delayHandler = context.getBean("delayerWithExpression.handler", DelayHandler.class);
assertNull(TestUtils.getPropertyValue(delayHandler, "delayHeaderName"));
assertEquals("100", TestUtils.getPropertyValue(delayHandler, "delayExpression", Expression.class).getExpressionString());
assertFalse(TestUtils.getPropertyValue(delayHandler, "ignoreExpressionFailures", Boolean.class));
}
@Test
public void testInt2243ExpressionSubElement() {
DelayHandler delayHandler = context.getBean("delayerWithExpressionSubElement.handler", DelayHandler.class);
assertNull(TestUtils.getPropertyValue(delayHandler, "delayHeaderName"));
assertEquals("headers.timestamp + 1000", TestUtils.getPropertyValue(delayHandler, "delayExpression", Expression.class).getExpressionString());
}
}

11
spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests-context.xml
View File

@@ -55,4 +55,15 @@
<beans:bean id="sampleHandler" class="org.springframework.integration.config.xml.DelayerUsageTests$SampleService"/>
<channel id="inputC"/>
<channel id="outputC">
<queue/>
</channel>
<beans:bean id="time" class="java.util.Calendar" factory-method="getInstance"/>
<delayer id="delayerExpression" input-channel="inputC" output-channel="outputC" expression="new java.util.Date(@time.timeInMillis + 5000)"/>
</beans:beans>

19
spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -56,6 +56,13 @@ public class DelayerUsageTests {
@Autowired @Qualifier("outputB1")
private PollableChannel outputB1;
@Autowired
private MessageChannel inputC;
@Autowired
private PollableChannel outputC;
@Test
public void testDelayWithDefaultScheduler(){
long start = System.currentTimeMillis();
@@ -109,6 +116,16 @@ public class DelayerUsageTests {
assertEquals("hello", message.getPayload());
}
@Test
public void testInt2243DelayerExpression() {
long start = System.currentTimeMillis();
this.inputC.send(new GenericMessage<String>("test"));
Message<?> message = this.outputC.receive(10000);
assertNotNull(message);
assertTrue((System.currentTimeMillis() - start) >= 1000);
assertEquals("test", message.getPayload());
}
public static class SampleService{
public String processMessage(String message) throws Exception {
Thread.sleep(500);

1
spring-integration-core/src/test/java/org/springframework/integration/config/xml/delayer-expression.properties Normal file
View File

@@ -0,0 +1 @@
delay=headers.timestamp + 1000

41
spring-integration-core/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java
View File

@@ -30,10 +30,14 @@ import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.context.support.StaticApplicationContext;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.Message;
import org.springframework.integration.MessageDeliveryException;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.context.IntegrationContextUtils;
@@ -50,6 +54,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
* @author Mark Fisher
* @author Artem Bilan
* @author Gunnar Hillert
* @author Gary Russell
* @since 1.0.3
*/
public class DelayHandlerTests {
@@ -80,6 +85,11 @@ public class DelayHandlerTests {
output.subscribe(resultHandler);
}
private void setDelayExpression() {
Expression expression = new SpelExpressionParser().parseExpression("headers.delay");
this.delayHandler.setDelayExpression(expression);
}
private void startDelayerHandler() {
delayHandler.afterPropertiesSet();
delayHandler.onApplicationEvent(new ContextRefreshedEvent(TestUtils.createTestApplicationContext()));
@@ -108,7 +118,7 @@ public class DelayHandlerTests {
@Test
public void delayHeaderAndDefaultDelayWouldTimeout() throws Exception {
delayHandler.setDefaultDelay(5000);
delayHandler.setDelayHeaderName("delay");
this.setDelayExpression();
this.startDelayerHandler();
Message<?> message = MessageBuilder.withPayload("test")
.setHeader("delay", 100).build();
@@ -121,7 +131,7 @@ public class DelayHandlerTests {
@Test
public void delayHeaderIsNegativeAndDefaultDelayWouldTimeout() throws Exception {
delayHandler.setDefaultDelay(5000);
delayHandler.setDelayHeaderName("delay");
this.setDelayExpression();
this.startDelayerHandler();
Message<?> message = MessageBuilder.withPayload("test")
.setHeader("delay", -7000).build();
@@ -134,7 +144,7 @@ public class DelayHandlerTests {
@Test
public void delayHeaderIsInvalidFallsBackToDefaultDelay() throws Exception {
delayHandler.setDefaultDelay(5);
delayHandler.setDelayHeaderName("delay");
this.setDelayExpression();
this.startDelayerHandler();
Message<?> message = MessageBuilder.withPayload("test")
.setHeader("delay", "not a number").build();
@@ -147,7 +157,7 @@ public class DelayHandlerTests {
@Test
public void delayHeaderIsDateInTheFutureAndDefaultDelayWouldTimeout() throws Exception {
delayHandler.setDefaultDelay(5000);
delayHandler.setDelayHeaderName("delay");
this.setDelayExpression();
this.startDelayerHandler();
Message<?> message = MessageBuilder.withPayload("test")
.setHeader("delay", new Date(new Date().getTime() + 150)).build();
@@ -160,7 +170,7 @@ public class DelayHandlerTests {
@Test
public void delayHeaderIsDateInThePastAndDefaultDelayWouldTimeout() throws Exception {
delayHandler.setDefaultDelay(5000);
delayHandler.setDelayHeaderName("delay");
this.setDelayExpression();
this.startDelayerHandler();
Message<?> message = MessageBuilder.withPayload("test")
.setHeader("delay", new Date(new Date().getTime() - 60 * 1000)).build();
@@ -172,7 +182,7 @@ public class DelayHandlerTests {
@Test
public void delayHeaderIsNullDateAndDefaultDelayIsZero() throws Exception {
delayHandler.setDelayHeaderName("delay");
this.setDelayExpression();
this.startDelayerHandler();
Date nullDate = null;
Message<?> message = MessageBuilder.withPayload("test")
@@ -185,7 +195,7 @@ public class DelayHandlerTests {
@Test(expected = TestTimedOutException.class)
public void delayHeaderIsFutureDateAndTimesOut() throws Exception {
delayHandler.setDelayHeaderName("delay");
this.setDelayExpression();
this.startDelayerHandler();
Date future = new Date(new Date().getTime() + 60 * 1000);
Message<?> message = MessageBuilder.withPayload("test")
@@ -199,7 +209,7 @@ public class DelayHandlerTests {
@Test
public void delayHeaderIsValidStringAndDefaultDelayWouldTimeout() throws Exception {
delayHandler.setDefaultDelay(5000);
delayHandler.setDelayHeaderName("delay");
this.setDelayExpression();
this.startDelayerHandler();
Message<?> message = MessageBuilder.withPayload("test")
.setHeader("delay", "20").build();
@@ -275,7 +285,7 @@ public class DelayHandlerTests {
MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler();
errorHandler.setDefaultErrorChannel(errorChannel);
taskScheduler.setErrorHandler(errorHandler);
delayHandler.setDelayHeaderName("delay");
this.setDelayExpression();
this.startDelayerHandler();
output.unsubscribe(resultHandler);
errorChannel.subscribe(resultHandler);
@@ -308,7 +318,7 @@ public class DelayHandlerTests {
MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler();
errorHandler.setBeanFactory(context);
taskScheduler.setErrorHandler(errorHandler);
delayHandler.setDelayHeaderName("delay");
this.setDelayExpression();
this.startDelayerHandler();
output.unsubscribe(resultHandler);
customErrorChannel.subscribe(resultHandler);
@@ -339,7 +349,7 @@ public class DelayHandlerTests {
MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler();
errorHandler.setBeanFactory(context);
taskScheduler.setErrorHandler(errorHandler);
delayHandler.setDelayHeaderName("delay");
this.setDelayExpression();
this.startDelayerHandler();
output.unsubscribe(resultHandler);
defaultErrorChannel.subscribe(resultHandler);
@@ -360,7 +370,6 @@ public class DelayHandlerTests {
assertNotSame(Thread.currentThread(), resultHandler.lastThread);
}
@Test //INT-1132
public void testReschedulePersistedMessagesOnStartup() throws Exception {
MessageGroupStore messageGroupStore = new SimpleMessageStore();
@@ -416,6 +425,14 @@ public class DelayHandlerTests {
Mockito.verify(this.delayHandler, Mockito.times(1)).reschedulePersistedMessages();
}
@Test(expected = MessageHandlingException.class)
public void testInt2243IgnoreExpressionFailuresAsFalse() throws Exception {
this.setDelayExpression();
this.delayHandler.setIgnoreExpressionFailures(false);
this.startDelayerHandler();
this.delayHandler.handleMessage(new GenericMessage<String>("test"));
}
private void waitForLatch(long timeout) {
try {
this.latch.await(timeout, TimeUnit.MILLISECONDS);

11
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/DelayerHandlerRescheduleIntegrationTests.java
View File

@@ -33,10 +33,10 @@ import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.handler.DelayHandler;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.util.UUIDConverter;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
@@ -75,7 +75,8 @@ public class DelayerHandlerRescheduleIntegrationTests {
MessageGroupStore messageStore = context.getBean("messageStore", MessageGroupStore.class);
assertEquals(0, messageStore.getMessageGroupCount());
input.send(MessageBuilder.withPayload("test1").build());
Message<String> message1 = MessageBuilder.withPayload("test1").build();
input.send(message1);
input.send(MessageBuilder.withPayload("test2").build());
// Emulate restart and check DB state before next start
@@ -101,8 +102,10 @@ public class DelayerHandlerRescheduleIntegrationTests {
MessageGroup messageGroup = messageStore.getMessageGroup(delayerMessageGroupId);
Message<?> messageInStore = messageGroup.getMessages().iterator().next();
Object payload = messageInStore.getPayload();
assertEquals("DelayedMessageWrapper", payload.getClass().getSimpleName());
assertEquals("test1", TestUtils.getPropertyValue(payload, "original.payload"));
//INT-3049
assertTrue(payload instanceof DelayHandler.DelayedMessageWrapper);
assertEquals(message1, ((DelayHandler.DelayedMessageWrapper) payload).getOriginal());
context.refresh();

57
src/reference/docbook/delayer.xml
View File

@@ -21,33 +21,56 @@
<para>
The <code>&lt;delayer&gt;</code> element is used to delay the Message flow between two Message Channels.
As with the other endpoints, you can provide the 'input-channel' and 'output-channel' attributes,
but the delayer also has 'default-delay' and 'delay-header-name' attributes that are used to
determine the number of milliseconds
that each Message should be delayed. The following delays all messages by 3 seconds:
but the delayer also has 'default-delay' and 'expression' attributes (and 'expression' sub-element) that are used to
determine the number of milliseconds that each Message should be delayed. The following delays all messages by 3 seconds:
<programlisting language="xml"><![CDATA[<int:delayer id="delayer" input-channel="input"
default-delay="3000" output-channel="output"/>]]></programlisting>
If you need per-Message determination of the delay, then you can also provide the name of a header
using the 'delay-header-name' attribute:
If you need per-Message determination of the delay, then you can also provide the SpEL expression
using the 'expression' attribute:
<programlisting language="xml"><![CDATA[<int:delayer id="delayer" input-channel="input" output-channel="output"
default-delay="3000" delay-header-name="delay"/>]]></programlisting>
In the example above the 3 second delay would only apply in the case that the header value is
not present for a given inbound Message. If you only want to apply a delay to Messages that have
an explicit header value, then you can set the 'default-delay' to 0 or don't use it at all (by default it is 0).
For any Message that has a delay of 0 (or less), the Message will be sent directly. In fact, if there is not a positive delay
value for a Message, it will be sent to the output channel on the calling Thread.
default-delay="3000" expression="headers['delay']"/>]]></programlisting>
In the example above, the 3 second delay would only apply when the expression evaluates to
<emphasis>null</emphasis> for a given inbound Message. If you only want to apply a delay to Messages that have
a valid result of the expression evaluation, then you can use a 'default-delay' of 0 (the default).
For any Message that has a delay of 0 (or less), the Message will be sent immediately, on the calling Thread.
<tip>
The delay handler supports header values that represent an interval in milliseconds (any
The delay handler supports expression evaluation results that represent an interval in milliseconds (any
Object whose <methodname>toString()</methodname> method produces a value that can be parsed into a
Long) as well as <classname>java.util.Date</classname> instances representing an absolute time.
In the first case, the milliseconds will be counted from the current time (e.g. a value of 5000
would delay the Message for at least 5 seconds from the time it is received by the Delayer).
With a Date instance, the Message will not be released until that Date
occurs. In either case, a value that equates to a non-positive delay, or a Date in the past, will
With a Date instance, the Message will not be released until the time represented by that Date object.
In either case, a value that equates to a non-positive delay, or a Date in the past, will
not result in any delay. Instead, it will be sent directly to the output channel on the original
sender's Thread. If the header is not a Date, and can not be parsed as a Long, the default
sender's Thread. If the expression evaluation result is not a Date, and can not be parsed as a Long, the default
delay (if any) will be applied.
</tip>
<important>
The expression evaluation may throw an evaluation Exception for various reasons, including an invalid
expression, or other conditions. By default, such exceptions are ignored (logged at DEBUG level) and
the delayer falls back to the default delay (if any). You can modify this behavior by setting the
<code>ignore-expression-failures</code> attribute.
By default this attribute is set to <code>true</code> and the Delayer behavior is as described above.
However, if you wish to not ignore expression evaluation exceptions, and throw them to the delayer's caller,
set the <code>ignore-expression-failures</code> attribute to <code>false</code>.
</important>
</para>
<tip>
Notice in the example above that the delay expression is specified as <code>headers['delay']</code>.
This is the SpEL <classname>Indexer</classname> syntax to access a <interfacename>Map</interfacename> element
(<classname>MessageHeaders</classname> implements <interfacename>Map</interfacename>),
it invokes: <code>headers.get("delay")</code>. For simple map element names (that do not contain '.')
you can also use the SpEL <emphasis>dot accessor</emphasis> syntax, where the above header expression
can be specified as <code>headers.delay</code>. But, different results are achieved if the header is missing.
In the first case, the expression will evaluate to <code>null</code>; the second will result in
something like:
<programlisting language="java"><![CDATA[ org.springframework.expression.spel.SpelEvaluationException: EL1008E:(pos 8):
Field or property 'delay' cannot be found on object of type 'org.springframework.integration.MessageHeaders']]></programlisting>
So, if there is a possibility of the header being omitted, and you want to fall back to the default
delay, it is generally more efficient (and recommended) to use the
<emphasis>Indexer</emphasis> syntax instead of <emphasis>dot property accessor</emphasis> syntax, because detecting
the null is faster than catching an exception.
</tip>
<para>
The delayer delegates to an instance of Spring's <interfacename>TaskScheduler</interfacename> abstraction.
The default scheduler used by the delayer is the <classname>ThreadPoolTaskScheduler</classname> instance
@@ -55,7 +78,7 @@
If you want to delegate to a different scheduler, you can provide a reference through the delayer element's
'scheduler' attribute:
<programlisting language="xml"><![CDATA[<int:delayer id="delayer" input-channel="input" output-channel="output"
delay-header-name="delay"
expression="headers.delay"
scheduler="exampleTaskScheduler"/>
<task:scheduler id="exampleTaskScheduler" pool-size="3"/>]]></programlisting>
@@ -106,7 +129,7 @@
<interfacename>org.aopalliance.aop.Advice</interfacename> implementation within the <code>&lt;advice-chain&gt;</code>.
A sample configuration of the <code>&lt;delayer&gt;</code> may look like this:
<programlisting language="xml"><![CDATA[<int:delayer id="delayer" input-channel="input" output-channel="output"
delay-header-name="delay"
expression="headers.delay"
message-store="jdbcMessageStore">
<int:advice-chain>
<beans:ref bean="customAdviceBean"/>

14
src/reference/docbook/whats-new.xml
View File

@@ -312,5 +312,19 @@
set <code>requires-reply</code> to false.
</important>
</section>
<section id="3.0-dalay-expression">
<title>Delayer: delay expression</title>
<para>
Previously, the <code>&lt;delayer&gt;</code> provided a <code>delay-header-name</code> attribute
to determine the <emphasis>delay</emphasis> value at runtime. In complex cases it was necessary
to precede the <code>&lt;delayer&gt;</code> with a <code>&lt;header-enricher&gt;</code>.
Spring Integration 3.0 introduced the <code>expression</code> attribute and <code>expression</code>
sub-element for dynamic delay determination. The <code>delay-header-name</code> attribute is now deprecated
because the header evaluation can be specified in the <code>expression</code>. In addition,
the <code>ignore-expression-failures</code> was introduced to control the behavior when an
expression evaluation fails.
For more information see <xref linkend="delayer"/>.
</para>
</section>
</section>
</chapter>