diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java index 428ff00671..109bb77f62 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/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; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/SubscriptionManager.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/SubscriptionManager.java index c8fcb6cb41..8c707c9d9b 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/SubscriptionManager.java +++ b/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; diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcherTask.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcherTask.java index 8b229b525a..6933727d75 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcherTask.java +++ b/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; diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/AbstractMessagingTaskScheduler.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/AbstractTaskScheduler.java similarity index 81% rename from org.springframework.integration/src/main/java/org/springframework/integration/scheduling/AbstractMessagingTaskScheduler.java rename to org.springframework.integration/src/main/java/org/springframework/integration/scheduling/AbstractTaskScheduler.java index d8a2898517..18da72d26f 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/AbstractMessagingTaskScheduler.java +++ b/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; diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/MessagingTask.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SchedulableTask.java similarity index 77% rename from org.springframework.integration/src/main/java/org/springframework/integration/scheduling/MessagingTask.java rename to org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SchedulableTask.java index 94314f2d40..1e1d3038f8 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/MessagingTask.java +++ b/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(); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/Schedule.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/Schedule.java index bbde1a13b6..f8074f616c 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/Schedule.java +++ b/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 */ diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SimpleTaskScheduler.java similarity index 64% rename from org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java rename to org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SimpleTaskScheduler.java index 91ddc053e8..34bf912299 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java +++ b/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 pendingTasks = new CopyOnWriteArraySet(); + private final Map> scheduledTasks = + new ConcurrentHashMap>(); + 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() + "'"); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/MessagingTaskScheduler.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/TaskScheduler.java similarity index 84% rename from org.springframework.integration/src/main/java/org/springframework/integration/scheduling/MessagingTaskScheduler.java rename to org.springframework.integration/src/main/java/org/springframework/integration/scheduling/TaskScheduler.java index d5974daad7..3d9822abef 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/MessagingTaskScheduler.java +++ b/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); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java index 7632a26934..459f45cb01 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java +++ b/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 diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java index e1d12634cb..4306021381 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java +++ b/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); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java index d661cad0b9..87dacab244 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java +++ b/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