Refactored MessagingTaskScheduler to support cancellation and renamed 'TaskScheduler'.

org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java

@@ -22,7 +22,6 @@ import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 
 import org.apache.commons.logging.Log;
@@ -63,10 +62,10 @@ import org.springframework.integration.message.CommandMessage;
 import org.springframework.integration.message.MessageTarget;
 import org.springframework.integration.message.PollCommand;
 import org.springframework.integration.scheduling.MessagePublishingErrorHandler;
-import org.springframework.integration.scheduling.MessagingTask;
-import org.springframework.integration.scheduling.MessagingTaskScheduler;
+import org.springframework.integration.scheduling.SchedulableTask;
+import org.springframework.integration.scheduling.TaskScheduler;
 import org.springframework.integration.scheduling.Schedule;
-import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler;
+import org.springframework.integration.scheduling.SimpleTaskScheduler;
 import org.springframework.integration.scheduling.Subscription;
 import org.springframework.util.Assert;
 
@@ -98,7 +97,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry,
 
 	private final MessageBusInterceptorsList interceptors = new MessageBusInterceptorsList();
 
-	private volatile MessagingTaskScheduler taskScheduler;
+	private volatile TaskScheduler taskScheduler;
 
 	private volatile boolean configureAsyncEventMulticaster = false;
 
@@ -137,10 +136,10 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry,
 	}
 
 	/**
-	 * Set the {@link MessagingTaskScheduler} to use for scheduling message dispatchers.
+	 * Set the {@link TaskScheduler} to use for scheduling message dispatchers.
 	 */
-	public void setMessagingTaskScheduler(MessagingTaskScheduler messagingTaskScheduler) {
-		this.taskScheduler = messagingTaskScheduler;
+	public void setTaskScheduler(TaskScheduler taskScheduler) {
+		this.taskScheduler = taskScheduler;
 	}
 
 	/**
@@ -205,7 +204,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry,
 			}
 			this.initializing = true;
 			if (this.taskScheduler == null) {
-				this.taskScheduler = new SimpleMessagingTaskScheduler(
+				this.taskScheduler = new SimpleTaskScheduler(
 						new ScheduledThreadPoolExecutor(DEFAULT_DISPATCHER_POOL_SIZE));
 			}
 			if (this.getErrorChannel() == null) {
@@ -389,7 +388,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry,
 		}
 		final Schedule schedule = endpoint.getSchedule();
 		if (schedule != null) {
-			this.taskScheduler.schedule(new MessagingTask() {
+			this.taskScheduler.schedule(new SchedulableTask() {
 				public Schedule getSchedule() {
 					return schedule;
 				}

org.springframework.integration/src/main/java/org/springframework/integration/bus/SubscriptionManager.java

@@ -31,7 +31,7 @@ import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.dispatcher.DirectChannel;
 import org.springframework.integration.dispatcher.PollingDispatcherTask;
 import org.springframework.integration.message.MessageTarget;
-import org.springframework.integration.scheduling.MessagingTaskScheduler;
+import org.springframework.integration.scheduling.TaskScheduler;
 import org.springframework.integration.scheduling.PollingSchedule;
 import org.springframework.integration.scheduling.Schedule;
 import org.springframework.util.Assert;
@@ -48,7 +48,7 @@ public class SubscriptionManager {
 
 	private final MessageChannel channel;
 
-	private final MessagingTaskScheduler scheduler;
+	private final TaskScheduler scheduler;
 
 	private volatile Schedule defaultSchedule = new PollingSchedule(5);
 
@@ -61,7 +61,7 @@ public class SubscriptionManager {
 	private final Object lifecycleMonitor = new Object();
 
 
-	public SubscriptionManager(MessageChannel channel, MessagingTaskScheduler scheduler) {
+	public SubscriptionManager(MessageChannel channel, TaskScheduler scheduler) {
 		Assert.notNull(channel, "channel must not be null");
 		Assert.notNull(scheduler, "scheduler must not be null");
 		this.channel = channel;

org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcherTask.java

@@ -20,16 +20,16 @@ import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.Subscribable;
 import org.springframework.integration.message.MessageTarget;
-import org.springframework.integration.scheduling.MessagingTask;
+import org.springframework.integration.scheduling.SchedulableTask;
 import org.springframework.integration.scheduling.Schedule;
 import org.springframework.util.Assert;
 
 /**
- * A {@link MessagingTask} that combines polling and dispatching.
+ * A {@link SchedulableTask} that combines polling and dispatching.
  * 
  * @author Mark Fisher
  */
-public class PollingDispatcherTask implements MessagingTask, Subscribable {
+public class PollingDispatcherTask implements SchedulableTask, Subscribable {
 
 	private final MessageChannel channel;
 

org.springframework.integration/src/main/java/org/springframework/integration/scheduling/AbstractTaskScheduler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2007 the original author or authors.
+ * Copyright 2002-2008 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.
@@ -19,11 +19,11 @@ package org.springframework.integration.scheduling;
 import java.util.concurrent.ScheduledFuture;
 
 /**
- * Base class for {@link MessagingTaskScheduler} implementations.
+ * Base class for {@link TaskScheduler} implementations.
  * 
  * @author Mark Fisher
  */
-public abstract class AbstractMessagingTaskScheduler implements MessagingTaskScheduler {
+public abstract class AbstractTaskScheduler implements TaskScheduler {
 
 	public boolean prefersShortLivedTasks() {
 		return true;

org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SchedulableTask.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2007 the original author or authors.
+ * Copyright 2002-2008 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,12 +17,11 @@
 package org.springframework.integration.scheduling;
 
 /**
- * An extension of runnable that provides metadata for a
- * {@link MessagingTaskScheduler}.
+ * An extension of runnable that provides metadata for a {@link TaskScheduler}.
  * 
  * @author Mark Fisher
  */
-public interface MessagingTask extends Runnable {
+public interface SchedulableTask extends Runnable {
 
 	Schedule getSchedule();
 

org.springframework.integration/src/main/java/org/springframework/integration/scheduling/Schedule.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2007 the original author or authors.
+ * Copyright 2002-2008 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,9 +17,9 @@
 package org.springframework.integration.scheduling;
 
 /**
- * A marker interface for scheduling metadata. Implementations of this interface
- * will provide the information necessary for a {@link MessagingTaskScheduler}
- * implementation to schedule tasks.
+ * A marker interface for scheduling metadata. Implementations
+ * of this interface will provide the information necessary for
+ * a {@link TaskScheduler} implementation to schedule tasks.
  * 
  * @author Mark Fisher
  */

org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SimpleTaskScheduler.java

@@ -16,7 +16,9 @@
 
 package org.springframework.integration.scheduling;
 
+import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
@@ -30,12 +32,13 @@ import org.springframework.integration.util.ErrorHandler;
 import org.springframework.util.Assert;
 
 /**
- * An implementation of {@link MessagingTaskScheduler} that understands
- * {@link PollingSchedule PollingSchedules}.
+ * An implementation of {@link TaskScheduler} that understands
+ * {@link PollingSchedule PollingSchedules} and delegates to
+ * a {@link ScheduledExecutorService} instance.
  * 
  * @author Mark Fisher
  */
-public class SimpleMessagingTaskScheduler extends AbstractMessagingTaskScheduler implements DisposableBean {
+public class SimpleTaskScheduler extends AbstractTaskScheduler implements DisposableBean {
 
 	private final Log logger = LogFactory.getLog(this.getClass());
 
@@ -47,12 +50,15 @@ public class SimpleMessagingTaskScheduler extends AbstractMessagingTaskScheduler
 
 	private final Set<Runnable> pendingTasks = new CopyOnWriteArraySet<Runnable>();
 
+	private final Map<Runnable, ScheduledFuture<?>> scheduledTasks =
+			new ConcurrentHashMap<Runnable, ScheduledFuture<?>>();
+
 	private volatile boolean running;
 
 	private final Object lifecycleMonitor = new Object();
 
 
-	public SimpleMessagingTaskScheduler(ScheduledExecutorService executor) {
+	public SimpleTaskScheduler(ScheduledExecutorService executor) {
 		Assert.notNull(executor, "'executor' must not be null");
 		this.executor = executor;
 	}
@@ -114,28 +120,47 @@ public class SimpleMessagingTaskScheduler extends AbstractMessagingTaskScheduler
 
 	@Override
 	public ScheduledFuture<?> schedule(Runnable task) {
-		if (!this.running) {
-			this.pendingTasks.add(task);
-			return null;
-		}
-		Schedule schedule = (task instanceof MessagingTask) ? ((MessagingTask) task).getSchedule() : null;
-		MessagingTaskRunner runner = new MessagingTaskRunner(task);
-		if (schedule == null) {
-			return this.executor.schedule(runner, 0, TimeUnit.MILLISECONDS);
-		}
-		if (schedule instanceof PollingSchedule) {
-			PollingSchedule ps = (PollingSchedule) schedule;
-			if (ps.getPeriod() <= 0) {
-				runner.setShouldRepeat(true);
-				return this.executor.schedule(runner, ps.getInitialDelay(), ps.getTimeUnit());
+		synchronized (this.lifecycleMonitor) {
+			if (!this.running) {
+				this.pendingTasks.add(task);
+				return null;
 			}
-			if (ps.getFixedRate()) {
-				return this.executor.scheduleAtFixedRate(runner, ps.getInitialDelay(), ps.getPeriod(), ps.getTimeUnit());
+			Schedule schedule = (task instanceof SchedulableTask) ? ((SchedulableTask) task).getSchedule() : null;
+			MessagingTaskRunner runner = new MessagingTaskRunner(task);
+			ScheduledFuture<?> future = null;
+			if (schedule == null) {
+				future = this.executor.schedule(runner, 0, TimeUnit.MILLISECONDS);
 			}
-			return this.executor.scheduleWithFixedDelay(runner, ps.getInitialDelay(), ps.getPeriod(), ps.getTimeUnit());
+			else if (schedule instanceof PollingSchedule) {
+				PollingSchedule ps = (PollingSchedule) schedule;
+				if (ps.getPeriod() <= 0) {
+					runner.setShouldRepeat(true);
+					future = this.executor.schedule(runner, ps.getInitialDelay(), ps.getTimeUnit());
+				}
+				else if (ps.getFixedRate()) {
+					future = this.executor.scheduleAtFixedRate(runner, ps.getInitialDelay(), ps.getPeriod(), ps.getTimeUnit());
+				}
+				else {
+					future = this.executor.scheduleWithFixedDelay(runner, ps.getInitialDelay(), ps.getPeriod(), ps.getTimeUnit());
+				}
+			}
+			if (future == null) {
+				throw new UnsupportedOperationException(this.getClass().getName() + " does not support schedule type '"
+						+ schedule.getClass().getName() + "'");
+			}
+			this.scheduledTasks.put(task, future);
+			return future;
+		}
+	}
+
+	public boolean cancel(Runnable task, boolean mayInterruptIfRunning) {
+		synchronized (this.lifecycleMonitor) {
+			ScheduledFuture<?> future = this.scheduledTasks.get(task);
+			if (future != null) {
+				return future.cancel(mayInterruptIfRunning);
+			}
+			return this.pendingTasks.remove(task);
 		}
-		throw new UnsupportedOperationException(this.getClass().getName() + " does not support schedule type '"
-				+ schedule.getClass().getName() + "'");
 	}
 
 

org.springframework.integration/src/main/java/org/springframework/integration/scheduling/TaskScheduler.java

@@ -23,14 +23,16 @@ import org.springframework.integration.util.ErrorHandler;
 import org.springframework.scheduling.SchedulingTaskExecutor;
 
 /**
- * Base interface for {@link MessagingTask} schedulers.
+ * Base interface for scheduling messaging tasks.
  * 
  * @author Mark Fisher
  */
-public interface MessagingTaskScheduler extends SchedulingTaskExecutor, Lifecycle {
+public interface TaskScheduler extends SchedulingTaskExecutor, Lifecycle {
 
 	ScheduledFuture<?> schedule(Runnable task);
 
+	boolean cancel(Runnable task, boolean mayInterruptIfRunning);
+
 	void setErrorHandler(ErrorHandler errorHandler);
 
 }

org.springframework.integration/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java

@@ -46,14 +46,14 @@ import org.springframework.integration.message.MessageTarget;
 import org.springframework.integration.message.StringMessage;
 import org.springframework.integration.message.selector.PayloadTypeSelector;
 import org.springframework.integration.scheduling.MessagePublishingErrorHandler;
-import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler;
+import org.springframework.integration.scheduling.SimpleTaskScheduler;
 
 /**
  * @author Mark Fisher
  */
 public class SubscriptionManagerTests {
 
-	private SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(new ScheduledThreadPoolExecutor(10));
+	private SimpleTaskScheduler scheduler = new SimpleTaskScheduler(new ScheduledThreadPoolExecutor(10));
 
 
 	@Test

org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java

@@ -40,7 +40,7 @@ import org.springframework.integration.message.MessageDeliveryException;
 import org.springframework.integration.message.MessagePriority;
 import org.springframework.integration.message.StringMessage;
 import org.springframework.integration.message.MessageTarget;
-import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler;
+import org.springframework.integration.scheduling.SimpleTaskScheduler;
 
 /**
  * @author Mark Fisher
@@ -70,7 +70,7 @@ public class ChannelParserTests {
 				"channelParserTests.xml", this.getClass());
 		MessageChannel channel = (MessageChannel) context.getBean("pointToPointChannelByDefault");
 		channel.send(new StringMessage("test"));
-		SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(new ScheduledThreadPoolExecutor(1));
+		SimpleTaskScheduler scheduler = new SimpleTaskScheduler(new ScheduledThreadPoolExecutor(1));
 		SubscriptionManager manager = new SubscriptionManager(channel, scheduler);
 		AtomicInteger counter = new AtomicInteger();
 		CountDownLatch latch = new CountDownLatch(1);
@@ -90,7 +90,7 @@ public class ChannelParserTests {
 				"channelParserTests.xml", this.getClass());
 		MessageChannel channel = (MessageChannel) context.getBean("pointToPointChannelExplicitlyConfigured");
 		channel.send(new StringMessage("test"));
-		SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(new ScheduledThreadPoolExecutor(1));
+		SimpleTaskScheduler scheduler = new SimpleTaskScheduler(new ScheduledThreadPoolExecutor(1));
 		SubscriptionManager manager = new SubscriptionManager(channel, scheduler);
 		AtomicInteger counter = new AtomicInteger();
 		CountDownLatch latch = new CountDownLatch(1);
@@ -110,7 +110,7 @@ public class ChannelParserTests {
 				"channelParserTests.xml", this.getClass());
 		MessageChannel channel = (MessageChannel) context.getBean("publishSubscribeChannel");
 		channel.send(new StringMessage("test"));
-		SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(new ScheduledThreadPoolExecutor(1));
+		SimpleTaskScheduler scheduler = new SimpleTaskScheduler(new ScheduledThreadPoolExecutor(1));
 		SubscriptionManager manager = new SubscriptionManager(channel, scheduler);
 		AtomicInteger counter = new AtomicInteger();
 		CountDownLatch latch = new CountDownLatch(2);

org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java

@@ -39,7 +39,7 @@ import org.springframework.integration.bus.interceptor.TestMessageBusStopInterce
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.dispatcher.DirectChannel;
 import org.springframework.integration.handler.TestHandlers;
-import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler;
+import org.springframework.integration.scheduling.SimpleTaskScheduler;
 import org.springframework.integration.scheduling.Subscription;
 
 /**
@@ -175,7 +175,7 @@ public class MessageBusParserTests {
 				context.getBean(AbstractApplicationContext.APPLICATION_EVENT_MULTICASTER_BEAN_NAME);
 		DirectFieldAccessor accessor = new DirectFieldAccessor(multicaster);
 		Object taskExecutor = accessor.getPropertyValue("taskExecutor");
-		assertEquals(SimpleMessagingTaskScheduler.class, taskExecutor.getClass());
+		assertEquals(SimpleTaskScheduler.class, taskExecutor.getClass());
 	}
 
 	@Test