INT-735 Added DelayHandler.
This commit is contained in:
Mark Fisher
@@ -0,0 +1,227 @@
|
||||
/*
|
||||
* 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 java.util.Date;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
import org.springframework.beans.BeansException;
|
||||
import org.springframework.beans.factory.BeanFactory;
|
||||
import org.springframework.beans.factory.BeanFactoryAware;
|
||||
import org.springframework.beans.factory.DisposableBean;
|
||||
import org.springframework.integration.channel.BeanFactoryChannelResolver;
|
||||
import org.springframework.integration.channel.ChannelResolutionException;
|
||||
import org.springframework.integration.channel.ChannelResolver;
|
||||
import org.springframework.integration.channel.MessageChannelTemplate;
|
||||
import org.springframework.integration.core.Message;
|
||||
import org.springframework.integration.core.MessageChannel;
|
||||
import org.springframework.integration.message.MessageHandler;
|
||||
import org.springframework.integration.scheduling.TaskScheduler;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* 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
|
||||
* 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
|
||||
* is that many delays can be managed concurrently, even very long delays,
|
||||
* without producing a buildup of blocked Threads.
|
||||
* <p>
|
||||
* 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.
|
||||
* <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
|
||||
* 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>).
|
||||
*
|
||||
* @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 <emphasis>only</emphasis> be applied to Messages with a
|
||||
* header, then set this value to 0.
|
||||
*/
|
||||
public void setDefaultDelay(long defaultDelay) {
|
||||
this.defaultDelay = defaultDelay;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
*/
|
||||
public void setDelayHeaderName(String delayHeaderName) {
|
||||
this.delayHeaderName = delayHeaderName;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the output channel for this handler. If none is provided, each
|
||||
* inbound Message must include a reply channel header.
|
||||
*/
|
||||
public void setOutputChannel(MessageChannel outputChannel) {
|
||||
this.outputChannel = outputChannel;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the timeout for sending reply Messages.
|
||||
*/
|
||||
public void setSendTimeout(long sendTimeout) {
|
||||
this.channelTemplate.setSendTimeout(sendTimeout);
|
||||
}
|
||||
|
||||
/**
|
||||
* Set whether to wait for scheduled tasks to complete on shutdown.
|
||||
* <p>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();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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");
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user