diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/DelayerParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/DelayerParser.java index 9c2ed01553..107674c015 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/DelayerParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/DelayerParser.java @@ -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"); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java index 696d0da20f..76ce4d64af 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java @@ -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. *

- * 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 headerDate.getTime() - new Date().getTime()). + * 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 { +public class DelayHandler extends AbstractReplyProducingMessageHandler implements DelayHandlerManagement, + ApplicationListener { + + 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 only be applied to Messages with a - * header, then set this value to 0. + * a delay should only 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 - null 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; diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd index 071527fdc8..84acd827e4 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd @@ -1359,8 +1359,8 @@ 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. @@ -1384,6 +1384,14 @@ to proxy DelayHandler's 'release Message task'. + + + + Specify the SpEL expression that evaluates to the delay value in milliseconds, + or a java.util.Date. + + + @@ -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. + + + + Specify the SpEL expression that evaluates to the delay value in milliseconds, + or a java.util.Date. + + + + + + + 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'. + + + - 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. diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerParserTests-context.xml index c1e6f0c928..95864237e5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerParserTests-context.xml @@ -57,6 +57,22 @@ + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerParserTests.java index 70f5a15f3a..3a5cd1a42c 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerParserTests.java @@ -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()); + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests-context.xml index b9e4ed48de..9136b78979 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests-context.xml @@ -55,4 +55,15 @@ + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests.java index 09cd1fed3a..3247f9eb8d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests.java @@ -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("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); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/delayer-expression.properties b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/delayer-expression.properties new file mode 100644 index 0000000000..2adde36dac --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/delayer-expression.properties @@ -0,0 +1 @@ +delay=headers.timestamp + 1000 diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java index 14ebd9587a..8238c63156 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java @@ -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("test")); + } + private void waitForLatch(long timeout) { try { this.latch.await(timeout, TimeUnit.MILLISECONDS); diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/DelayerHandlerRescheduleIntegrationTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/DelayerHandlerRescheduleIntegrationTests.java index f2f75f0cc4..1a4ed4ae94 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/DelayerHandlerRescheduleIntegrationTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/DelayerHandlerRescheduleIntegrationTests.java @@ -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 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(); diff --git a/src/reference/docbook/delayer.xml b/src/reference/docbook/delayer.xml index c750983211..7d2847757c 100644 --- a/src/reference/docbook/delayer.xml +++ b/src/reference/docbook/delayer.xml @@ -21,33 +21,56 @@ The <delayer> 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: ]]> - 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: ]]> - 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']"/>]]> + In the example above, the 3 second delay would only apply when the expression evaluates to + null 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. - 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 toString() method produces a value that can be parsed into a Long) as well as java.util.Date 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. + + 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 + ignore-expression-failures attribute. + By default this attribute is set to true 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 ignore-expression-failures attribute to false. + + + Notice in the example above that the delay expression is specified as headers['delay']. + This is the SpEL Indexer syntax to access a Map element + (MessageHeaders implements Map), + it invokes: headers.get("delay"). For simple map element names (that do not contain '.') + you can also use the SpEL dot accessor syntax, where the above header expression + can be specified as headers.delay. But, different results are achieved if the header is missing. + In the first case, the expression will evaluate to null; the second will result in + something like: + + 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 + Indexer syntax instead of dot property accessor syntax, because detecting + the null is faster than catching an exception. + The delayer delegates to an instance of Spring's TaskScheduler abstraction. The default scheduler used by the delayer is the ThreadPoolTaskScheduler 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: ]]> @@ -106,7 +129,7 @@ org.aopalliance.aop.Advice implementation within the <advice-chain>. A sample configuration of the <delayer> may look like this: diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index c4ea63eee9..8013df27cd 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -312,5 +312,19 @@ set requires-reply to false. +

+ Delayer: delay expression + + Previously, the <delayer> provided a delay-header-name attribute + to determine the delay value at runtime. In complex cases it was necessary + to precede the <delayer> with a <header-enricher>. + Spring Integration 3.0 introduced the expression attribute and expression + sub-element for dynamic delay determination. The delay-header-name attribute is now deprecated + because the header evaluation can be specified in the expression. In addition, + the ignore-expression-failures was introduced to control the behavior when an + expression evaluation fails. + For more information see . + +