+ * One thing to keep in mind, however, is that any active transactional context + * will not propagate from the original sender to the eventual recipient. This + * 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
+ * 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()).
+ *
+ * @author Mark Fisher
+ * @since 1.0.3
+ */
+public class DelayHandler implements MessageHandler, BeanFactoryAware, DisposableBean {
+
+ private final Log logger = LogFactory.getLog(this.getClass());
+
+ private volatile long defaultDelay;
+
+ private volatile String delayHeaderName;
+
+ private final ScheduledExecutorService scheduler;
+
+ private volatile MessageChannel outputChannel;
+
+ private volatile ChannelResolver channelResolver;
+
+ private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate();
+
+ private volatile boolean waitForTasksToCompleteOnShutdown;
+
+
+ /**
+ * Create a DelayHandler with the given default delay. The sending of Messages after
+ * the delay will be handled by a scheduled thread pool with a size of 1.
+ */
+ public DelayHandler(long defaultDelay) {
+ this(defaultDelay, null);
+ }
+
+ /**
+ * Create a DelayHandler with the given default delay. The sending of Messages after
+ * the delay will be handled by the provided {@link ScheduledExecutorService}.
+ */
+ public DelayHandler(long defaultDelay, ScheduledExecutorService scheduledExecutorService) {
+ this.defaultDelay = defaultDelay;
+ this.scheduler = (scheduledExecutorService != null)
+ ? scheduledExecutorService : Executors.newScheduledThreadPool(1);
+ }
+
+
+ /**
+ * Set the default delay in milliseconds. If no 'delayHeaderName' property
+ * has been provided, the default delay will be applied to all Messages. If
+ * a delay should
Default is "false". Switch this to "true" if you prefer + * fully completed tasks at the expense of a longer shutdown phase. + * @see java.util.concurrent.ExecutorService#shutdown() + * @see java.util.concurrent.ExecutorService#shutdownNow() + */ + public void setWaitForTasksToCompleteOnShutdown(boolean waitForJobsToCompleteOnShutdown) { + this.waitForTasksToCompleteOnShutdown = waitForJobsToCompleteOnShutdown; + } + + public void setBeanFactory(BeanFactory beanFactory) throws BeansException { + this.channelResolver = new BeanFactoryChannelResolver(beanFactory); + } + + public final void handleMessage(final 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(); + } + else if (headerValue != null) { + try { + delay = Long.valueOf(headerValue.toString()); + } + catch (NumberFormatException e) { + if (logger.isDebugEnabled()) { + logger.debug("Failed to parse delay from header value '" + headerValue.toString() + + "', will fall back to default delay: " + this.defaultDelay); + } + } + } + } + if (delay > 0) { + this.scheduler.schedule(new Runnable() { + public void run() { + releaseMessage(message); + } + }, delay, TimeUnit.MILLISECONDS); + } + else { + // no delay, release directly + this.releaseMessage(message); + } + } + + private void releaseMessage(Message message) { + MessageChannel replyChannel = this.resolveReplyChannel(message); + this.channelTemplate.send(message, replyChannel); + } + + private MessageChannel resolveReplyChannel(Message message) { + MessageChannel replyChannel = this.outputChannel; + if (replyChannel == null) { + Object replyChannelHeader= message.getHeaders().getReplyChannel(); + if (replyChannelHeader != null) { + if (replyChannelHeader instanceof MessageChannel) { + replyChannel = (MessageChannel) replyChannelHeader; + } + else if (replyChannelHeader instanceof String) { + Assert.state(this.channelResolver != null, + "ChannelResolver is required for resolving a reply channel by name"); + replyChannel = this.channelResolver.resolveChannelName((String) replyChannelHeader); + } + else { + throw new ChannelResolutionException("expected a MessageChannel or String for 'replyChannel', but type is [" + + replyChannelHeader.getClass() + "]"); + } + } + } + if (replyChannel == null) { + throw new ChannelResolutionException( + "unable to resolve reply channel for message: " + message); + } + return replyChannel; + } + + public void destroy() { + if (this.waitForTasksToCompleteOnShutdown) { + this.scheduler.shutdown(); + } + else { + this.scheduler.shutdownNow(); + } + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java new file mode 100644 index 0000000000..75af651edc --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java @@ -0,0 +1,290 @@ +/* + * Copyright 2002-2009 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.handler; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertSame; + +import java.util.Date; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +import org.junit.Test; + +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.core.Message; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.message.MessageHandler; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + * @since 1.0.3 + */ +public class DelayHandlerTests { + + private final DirectChannel input = new DirectChannel(); + + private final DirectChannel output = new DirectChannel(); + + private final CountDownLatch latch = new CountDownLatch(1); + + + @Test + public void noDelayHeaderAndDefaultDelayIsZero() { + DelayHandler delayHandler = new DelayHandler(0); + ResultHandler resultHandler = new ResultHandler(); + delayHandler.setOutputChannel(output); + input.subscribe(delayHandler); + output.subscribe(resultHandler); + Message message = MessageBuilder.withPayload("test").build(); + input.send(message); + assertSame(message, resultHandler.lastMessage); + assertSame(Thread.currentThread(), resultHandler.lastThread); + } + + @Test + public void noDelayHeaderAndDefaultDelayIsPositive() { + DelayHandler delayHandler = new DelayHandler(10); + ResultHandler resultHandler = new ResultHandler(); + delayHandler.setOutputChannel(output); + input.subscribe(delayHandler); + output.subscribe(resultHandler); + Message message = MessageBuilder.withPayload("test").build(); + input.send(message); + this.waitForLatch(1000); + assertSame(message, resultHandler.lastMessage); + assertNotSame(Thread.currentThread(), resultHandler.lastThread); + } + + @Test + public void delayHeaderAndDefaultDelayWouldTimeout() { + DelayHandler delayHandler = new DelayHandler(5000); + delayHandler.setDelayHeaderName("delay"); + ResultHandler resultHandler = new ResultHandler(); + delayHandler.setOutputChannel(output); + input.subscribe(delayHandler); + output.subscribe(resultHandler); + Message message = MessageBuilder.withPayload("test") + .setHeader("delay", 100).build(); + input.send(message); + this.waitForLatch(1000); + assertSame(message, resultHandler.lastMessage); + assertNotSame(Thread.currentThread(), resultHandler.lastThread); + } + + @Test + public void delayHeaderIsNegativeAndDefaultDelayWouldTimeout() { + DelayHandler delayHandler = new DelayHandler(5000); + delayHandler.setDelayHeaderName("delay"); + ResultHandler resultHandler = new ResultHandler(); + delayHandler.setOutputChannel(output); + input.subscribe(delayHandler); + output.subscribe(resultHandler); + Message message = MessageBuilder.withPayload("test") + .setHeader("delay", -7000).build(); + input.send(message); + this.waitForLatch(1000); + assertSame(message, resultHandler.lastMessage); + assertSame(Thread.currentThread(), resultHandler.lastThread); + } + + @Test + public void delayHeaderIsInvalidFallsBackToDefaultDelay() { + DelayHandler delayHandler = new DelayHandler(5); + delayHandler.setDelayHeaderName("delay"); + ResultHandler resultHandler = new ResultHandler(); + delayHandler.setOutputChannel(output); + input.subscribe(delayHandler); + output.subscribe(resultHandler); + Message message = MessageBuilder.withPayload("test") + .setHeader("delay", "not a number").build(); + input.send(message); + this.waitForLatch(1000); + assertSame(message, resultHandler.lastMessage); + assertNotSame(Thread.currentThread(), resultHandler.lastThread); + } + + @Test + public void delayHeaderIsDateInTheFutureAndDefaultDelayWouldTimeout() { + DelayHandler delayHandler = new DelayHandler(5000); + delayHandler.setDelayHeaderName("delay"); + ResultHandler resultHandler = new ResultHandler(); + delayHandler.setOutputChannel(output); + input.subscribe(delayHandler); + output.subscribe(resultHandler); + Message message = MessageBuilder.withPayload("test") + .setHeader("delay", new Date(new Date().getTime() + 150)).build(); + input.send(message); + this.waitForLatch(3000); + assertSame(message, resultHandler.lastMessage); + assertNotSame(Thread.currentThread(), resultHandler.lastThread); + } + + @Test + public void delayHeaderIsDateInThePastAndDefaultDelayWouldTimeout() { + DelayHandler delayHandler = new DelayHandler(5000); + delayHandler.setDelayHeaderName("delay"); + ResultHandler resultHandler = new ResultHandler(); + delayHandler.setOutputChannel(output); + input.subscribe(delayHandler); + output.subscribe(resultHandler); + Message message = MessageBuilder.withPayload("test") + .setHeader("delay", new Date(new Date().getTime() - 60 * 1000)).build(); + input.send(message); + this.waitForLatch(3000); + assertSame(message, resultHandler.lastMessage); + assertSame(Thread.currentThread(), resultHandler.lastThread); + } + + @Test + public void delayHeaderIsNullDateAndDefaultDelayIsZero() { + DelayHandler delayHandler = new DelayHandler(0); + delayHandler.setDelayHeaderName("delay"); + ResultHandler resultHandler = new ResultHandler(); + delayHandler.setOutputChannel(output); + input.subscribe(delayHandler); + output.subscribe(resultHandler); + Date nullDate = null; + Message message = MessageBuilder.withPayload("test") + .setHeader("delay", nullDate).build(); + input.send(message); + this.waitForLatch(3000); + assertSame(message, resultHandler.lastMessage); + assertSame(Thread.currentThread(), resultHandler.lastThread); + } + + @Test(expected = TestTimedOutException.class) + public void delayHeaderIsFutureDateAndTimesOut() { + DelayHandler delayHandler = new DelayHandler(0); + delayHandler.setDelayHeaderName("delay"); + ResultHandler resultHandler = new ResultHandler(); + delayHandler.setOutputChannel(output); + input.subscribe(delayHandler); + output.subscribe(resultHandler); + Date future = new Date(new Date().getTime() + 60 * 1000); + Message message = MessageBuilder.withPayload("test") + .setHeader("delay", future).build(); + input.send(message); + this.waitForLatch(50); + assertSame(message, resultHandler.lastMessage); + assertSame(Thread.currentThread(), resultHandler.lastThread); + } + + @Test + public void delayHeaderIsValidStringAndDefaultDelayWouldTimeout() { + DelayHandler delayHandler = new DelayHandler(5000); + delayHandler.setDelayHeaderName("delay"); + ResultHandler resultHandler = new ResultHandler(); + delayHandler.setOutputChannel(output); + input.subscribe(delayHandler); + output.subscribe(resultHandler); + Message message = MessageBuilder.withPayload("test") + .setHeader("delay", "20").build(); + input.send(message); + this.waitForLatch(1000); + assertSame(message, resultHandler.lastMessage); + assertNotSame(Thread.currentThread(), resultHandler.lastThread); + } + + @Test + public void verifyShutdownWithoutWaitingByDefault() throws InterruptedException { + DelayHandler delayHandler = new DelayHandler(5000); + delayHandler.handleMessage(new StringMessage("foo")); + delayHandler.destroy(); + final ScheduledThreadPoolExecutor scheduler = (ScheduledThreadPoolExecutor) + new DirectFieldAccessor(delayHandler).getPropertyValue("scheduler"); + final CountDownLatch latch = new CountDownLatch(1); + new Thread(new Runnable() { + public void run() { + try { + scheduler.awaitTermination(10000, TimeUnit.MILLISECONDS); + latch.countDown(); + } + catch (InterruptedException e) { + // won't countDown + } + } + }).start(); + latch.await(50, TimeUnit.MILLISECONDS); + assertEquals(0, latch.getCount()); + } + + @Test + public void verifyShutdownWithWait() throws InterruptedException { + DelayHandler delayHandler = new DelayHandler(5000); + delayHandler.setWaitForTasksToCompleteOnShutdown(true); + delayHandler.handleMessage(new StringMessage("foo")); + delayHandler.destroy(); + final ScheduledThreadPoolExecutor scheduler = (ScheduledThreadPoolExecutor) + new DirectFieldAccessor(delayHandler).getPropertyValue("scheduler"); + final CountDownLatch latch = new CountDownLatch(1); + new Thread(new Runnable() { + public void run() { + try { + scheduler.awaitTermination(10000, TimeUnit.MILLISECONDS); + latch.countDown(); + } + catch (InterruptedException e) { + // won't countDown + } + } + }).start(); + latch.await(50, TimeUnit.MILLISECONDS); + assertEquals(1, latch.getCount()); + } + + + private void waitForLatch(long timeout) { + try { + this.latch.await(timeout, TimeUnit.MILLISECONDS); + if (latch.getCount() != 0) { + throw new TestTimedOutException(); + } + } + catch (InterruptedException e) { + throw new RuntimeException("interrupted while waiting for latch"); + } + } + + + private class ResultHandler implements MessageHandler { + + private volatile Message lastMessage; + + private volatile Thread lastThread; + + public void handleMessage(Message message) { + this.lastMessage = message; + this.lastThread = Thread.currentThread(); + latch.countDown(); + } + } + + + @SuppressWarnings("serial") + private static class TestTimedOutException extends RuntimeException { + + public TestTimedOutException() { + super("timed out while waiting for latch"); + } + } + +}