INT-737 DelayHandler now delegates to a configurable MessageStore for storing any Messages awaiting a delayed release.

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

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

org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd

@@ -928,6 +928,20 @@
 							</xsd:appinfo>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="message-store" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+								Provide a reference to the MessageStore instance that should be used
+								to store Messages while awaiting the delay.
+							</xsd:documentation>
+							<xsd:appinfo>
+								<tool:annotation kind="ref">
+									<tool:expected-type
+										type="org.springframework.integration.store.MessageStore" />
+								</tool:annotation>
+							</xsd:appinfo>
+						</xsd:annotation>
+					</xsd:attribute>
 					<xsd:attribute name="send-timeout" type="xsd:string">
 						<xsd:annotation>
 							<xsd:documentation>

org.springframework.integration/src/test/java/org/springframework/integration/config/xml/DelayerParserTests-context.xml

@@ -31,6 +31,14 @@
 			 default-delay="0"
 			 scheduler="testScheduler"/>
 
+	<delayer id="delayerWithCustomMessageStore"
+			 input-channel="input"
+			 output-channel="output"
+			 default-delay="0"
+			 message-store="testMessageStore"/>
+
 	<task:scheduler id="testScheduler" pool-size="7"/>
 
+	<beans:bean id="testMessageStore" class="org.springframework.integration.store.SimpleMessageStore"/>
+
 </beans:beans>

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