diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/DelayerParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/DelayerParser.java index c69f6981dd..037be313b5 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/DelayerParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/DelayerParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2009 the original author or authors. + * Copyright 2002-2010 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. @@ -44,6 +44,7 @@ public class DelayerParser extends AbstractConsumerEndpointParser { if (StringUtils.hasText(scheduler)) { builder.addConstructorArgReference(scheduler); } + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-store"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "delay-header-name"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "wait-for-tasks-to-complete-on-shutdown"); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/handler/DelayHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/handler/DelayHandler.java index 59acb9d7dd..20982caa7d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/handler/DelayHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/handler/DelayHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2009 the original author or authors. + * Copyright 2002-2010 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. @@ -17,6 +17,7 @@ package org.springframework.integration.handler; import java.util.Date; +import java.util.UUID; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -38,6 +39,8 @@ import org.springframework.integration.core.MessageHeaders; import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.message.MessageHandler; import org.springframework.integration.message.MessageHandlingException; +import org.springframework.integration.store.MessageStore; +import org.springframework.integration.store.SimpleMessageStore; import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.concurrent.ExecutorConfigurationSupport; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; @@ -84,6 +87,8 @@ public class DelayHandler implements MessageHandler, Ordered, BeanFactoryAware, private volatile ChannelResolver channelResolver; + private volatile MessageStore messageStore; + private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate(); private volatile int order = Ordered.LOWEST_PRECEDENCE; @@ -127,6 +132,14 @@ public class DelayHandler implements MessageHandler, Ordered, BeanFactoryAware, this.delayHeaderName = delayHeaderName; } + /** + * Specify the {@link MessageStore} that should be used to store Messages + * while awaiting the delay. + */ + public void setMessageStore(MessageStore messageStore) { + this.messageStore = messageStore; + } + /** * Set the output channel for this handler. If none is provided, each * inbound Message must include a reply channel header. @@ -173,7 +186,11 @@ public class DelayHandler implements MessageHandler, Ordered, BeanFactoryAware, this.channelResolver = new BeanFactoryChannelResolver(beanFactory); } + public void afterPropertiesSet() throws Exception { + if (this.messageStore == null) { + this.messageStore = new SimpleMessageStore(); + } if (this.taskScheduler instanceof InitializingBean) { ((InitializingBean) this.taskScheduler).afterPropertiesSet(); } @@ -185,8 +202,8 @@ public class DelayHandler implements MessageHandler, Ordered, BeanFactoryAware, this.releaseMessageAfterDelay(message, delay); } else { - // no delay, release directly - this.releaseMessage(message); + // no delay, send directly + this.sendMessageToReplyChannel(message); } } @@ -213,10 +230,12 @@ public class DelayHandler implements MessageHandler, Ordered, BeanFactoryAware, } private void releaseMessageAfterDelay(final Message message, long delay) { + Assert.state(this.messageStore != null, "MessageStore must not be null"); + final Message storedMessage = this.messageStore.addMessage(message); this.taskScheduler.schedule(new Runnable() { public void run() { try { - releaseMessage(message); + releaseMessage(storedMessage.getHeaders().getId()); } catch (Exception e) { Exception exception = new MessageHandlingException(message, "Failed to deliver Message after delay.", e); @@ -236,7 +255,14 @@ public class DelayHandler implements MessageHandler, Ordered, BeanFactoryAware, }, new Date(System.currentTimeMillis() + delay)); } - private void releaseMessage(Message message) { + private void releaseMessage(UUID id) { + Assert.state(this.messageStore != null, "MessageStore must not be null"); + Message message = this.messageStore.removeMessage(id); + Assert.notNull(message, "Message with id: " + id + " no longer exists in MessageStore."); + this.sendMessageToReplyChannel(message); + } + + private void sendMessageToReplyChannel(Message message) { MessageChannel replyChannel = this.resolveReplyChannel(message); this.channelTemplate.send(message, replyChannel); } diff --git a/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd b/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd index 41b02408ab..d3b42a24f7 100644 --- a/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd +++ b/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd @@ -928,6 +928,20 @@ + + + + Provide a reference to the MessageStore instance that should be used + to store Messages while awaiting the delay. + + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/DelayerParserTests-context.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/DelayerParserTests-context.xml index 3ca64320b5..cecafcd54b 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/DelayerParserTests-context.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/DelayerParserTests-context.xml @@ -31,6 +31,14 @@ default-delay="0" scheduler="testScheduler"/> + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/DelayerParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/DelayerParserTests.java index 31b3cac772..64f4703a9a 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/DelayerParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/DelayerParserTests.java @@ -61,7 +61,6 @@ public class DelayerParserTests { accessor.getPropertyValue("taskScheduler")).getPropertyValue("waitForTasksToCompleteOnShutdown")); } - @Test public void customScheduler() { Object endpoint = context.getBean("delayerWithCustomScheduler"); @@ -76,4 +75,15 @@ public class DelayerParserTests { assertEquals(context.getBean("testScheduler"), accessor.getPropertyValue("taskScheduler")); } + @Test + public void customMessageStore() { + Object endpoint = context.getBean("delayerWithCustomMessageStore"); + assertEquals(EventDrivenConsumer.class, endpoint.getClass()); + Object handler = TestUtils.getPropertyValue(endpoint, "handler"); + assertEquals(DelayHandler.class, handler.getClass()); + DelayHandler delayHandler = (DelayHandler) handler; + DirectFieldAccessor accessor = new DirectFieldAccessor(delayHandler); + assertEquals(context.getBean("testMessageStore"), accessor.getPropertyValue("messageStore")); + } + }