diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java index a12100d426..ec9d4321e4 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java @@ -26,7 +26,6 @@ import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.beans.BeansException; import org.springframework.beans.factory.DisposableBean; import org.springframework.context.ApplicationContext; @@ -60,8 +59,9 @@ import org.springframework.integration.message.SubscribableSource; import org.springframework.integration.scheduling.MessagePublishingErrorHandler; import org.springframework.integration.scheduling.PollingSchedule; import org.springframework.integration.scheduling.Schedule; -import org.springframework.integration.scheduling.SimpleTaskScheduler; import org.springframework.integration.scheduling.TaskScheduler; +import org.springframework.integration.scheduling.spi.ProviderTaskScheduler; +import org.springframework.integration.scheduling.spi.SimpleScheduleServiceProvider; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.util.Assert; @@ -210,7 +210,7 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(DEFAULT_DISPATCHER_POOL_SIZE); executor.setThreadFactory(new CustomizableThreadFactory("message-bus-")); executor.setRejectedExecutionHandler(new CallerRunsPolicy()); - this.taskScheduler = new SimpleTaskScheduler(executor); + this.taskScheduler = new ProviderTaskScheduler(new SimpleScheduleServiceProvider(executor)); } if (this.getErrorChannel() == null) { this.setErrorChannel(new DefaultErrorChannel()); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/AbstractTaskScheduler.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/AbstractTaskScheduler.java deleted file mode 100644 index 18da72d26f..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/AbstractTaskScheduler.java +++ /dev/null @@ -1,41 +0,0 @@ -/* - * 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. - * 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.scheduling; - -import java.util.concurrent.ScheduledFuture; - -/** - * Base class for {@link TaskScheduler} implementations. - * - * @author Mark Fisher - */ -public abstract class AbstractTaskScheduler implements TaskScheduler { - - public boolean prefersShortLivedTasks() { - return true; - } - - /** - * Submit a task to be run once. - */ - public void execute(Runnable task) { - this.schedule(task); - } - - public abstract ScheduledFuture schedule(Runnable task); - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/TaskScheduler.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/TaskScheduler.java index 3d9822abef..208650398c 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/TaskScheduler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/TaskScheduler.java @@ -29,7 +29,7 @@ import org.springframework.scheduling.SchedulingTaskExecutor; */ public interface TaskScheduler extends SchedulingTaskExecutor, Lifecycle { - ScheduledFuture schedule(Runnable task); + ScheduledFuture schedule(SchedulableTask task); boolean cancel(Runnable task, boolean mayInterruptIfRunning); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/spi/ProviderTaskScheduler.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/spi/ProviderTaskScheduler.java new file mode 100644 index 0000000000..034cb50131 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/spi/ProviderTaskScheduler.java @@ -0,0 +1,229 @@ +/* + * 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. + * 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.scheduling.spi; + +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArraySet; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.integration.scheduling.PollingSchedule; +import org.springframework.integration.scheduling.SchedulableTask; +import org.springframework.integration.scheduling.TaskScheduler; +import org.springframework.integration.util.ErrorHandler; +import org.springframework.util.Assert; + +/** + * An implementation of {@link org.springframework.integration.scheduling.TaskScheduler} that understands + * {@link org.springframework.integration.scheduling.PollingSchedule PollingSchedules} and delegates to + * a {@link ScheduleServiceProvider} instance. + * + * @author Mark Fisher + * @author Marius Bogoevici + */ +public class ProviderTaskScheduler implements TaskScheduler, DisposableBean { + + private final Log logger = LogFactory.getLog(this.getClass()); + + private ScheduleServiceProvider scheduleServiceProvider; + + private volatile boolean waitForTasksToCompleteOnShutdown = true; + + private volatile ErrorHandler errorHandler; + + private final Set pendingTasks = new CopyOnWriteArraySet(); + + private final Map> scheduledTasks = + new ConcurrentHashMap>(); + + private volatile boolean running; + + private final Object lifecycleMonitor = new Object(); + + + public ProviderTaskScheduler(ScheduleServiceProvider scheduleServiceProvider) { + Assert.notNull(scheduleServiceProvider, "'executor' must not be null"); + this.scheduleServiceProvider = scheduleServiceProvider; + } + + + public void setWaitForTasksToCompleteOnShutdown(boolean waitForTasksToCompleteOnShutdown) { + this.waitForTasksToCompleteOnShutdown = waitForTasksToCompleteOnShutdown; + } + + public void setErrorHandler(ErrorHandler errorHandler) { + this.errorHandler = errorHandler; + } + + public boolean isRunning() { + return this.running; + } + + public void start() { + synchronized (this.lifecycleMonitor) { + if (this.running) { + return; + } + if (logger.isInfoEnabled()) { + logger.info("task scheduler starting"); + } + this.running = true; + for (Runnable task : this.pendingTasks) { + if (task instanceof SchedulableTask) { + this.schedule((SchedulableTask)task); + } + else { + this.execute(task); + } + } + this.pendingTasks.clear(); + if (logger.isInfoEnabled()) { + logger.info("task scheduler started successfully"); + } + } + } + + public void stop() { + synchronized (this.lifecycleMonitor) { + if (this.running) { + if (logger.isInfoEnabled()) { + logger.info("task scheduler stopping"); + } + this.running = false; + for (Runnable task : this.scheduledTasks.keySet()) { + this.cancel(task, true); + } + if (logger.isInfoEnabled()) { + logger.info("task scheduler stopped successfully"); + } + } + } + } + + public void destroy() { + synchronized (this.lifecycleMonitor) { + this.stop(); + scheduleServiceProvider.shutdown(this.waitForTasksToCompleteOnShutdown); + } + } + + public boolean prefersShortLivedTasks() { + return true; + } + + public void execute(Runnable task) { + this.scheduleServiceProvider.execute(task); + } + + public ScheduledFuture schedule(SchedulableTask task) { + synchronized (this.lifecycleMonitor) { + if (!this.running) { + if (logger.isDebugEnabled()) { + logger.debug("scheduler is not running, adding task to pending list: " + task); + } + this.pendingTasks.add(task); + return null; + } + if (logger.isDebugEnabled()) { + logger.debug("scheduling task: " + task); + } + TaskRunner runner = new TaskRunner(task); + ScheduledFuture future = null; + if (task.getSchedule() == null) { + future = this.scheduleServiceProvider.scheduleWithInitialDelay(runner, 0, TimeUnit.MILLISECONDS); + } + else if (task.getSchedule() instanceof PollingSchedule) { + PollingSchedule ps = (PollingSchedule) task.getSchedule(); + if (ps.getPeriod() <= 0) { + runner.setShouldRepeat(true); + future = this.scheduleServiceProvider.scheduleWithInitialDelay(runner, ps.getInitialDelay(), ps.getTimeUnit()); + } + else if (ps.getFixedRate()) { + future = this.scheduleServiceProvider.scheduleAtFixedRate(runner, ps.getInitialDelay(), ps.getPeriod(), ps.getTimeUnit()); + } + else { + future = this.scheduleServiceProvider.scheduleWithFixedDelay(runner, ps.getInitialDelay(), ps.getPeriod(), ps.getTimeUnit()); + } + } + if (future == null) { + throw new UnsupportedOperationException(this.getClass().getName() + " does not support scheduleWithInitialDelay type '" + + task.getSchedule().getClass().getName() + "'"); + } + this.scheduledTasks.put(task, future); + if (logger.isDebugEnabled()) { + logger.debug("scheduled task: " + task); + } + return future; + } + } + + public boolean cancel(Runnable task, boolean mayInterruptIfRunning) { + synchronized (this.lifecycleMonitor) { + ScheduledFuture future = this.scheduledTasks.get(task); + if (future != null) { + if (logger.isDebugEnabled()) { + logger.debug("cancelling task: " + task); + } + return future.cancel(mayInterruptIfRunning); + } + return this.pendingTasks.remove(task); + } + } + + + private class TaskRunner implements Runnable { + + private final Runnable task; + + private volatile boolean shouldRepeat; + + + public TaskRunner(Runnable task) { + this.task = task; + } + + + public void setShouldRepeat(boolean shouldRepeat) { + this.shouldRepeat = shouldRepeat; + } + + public void run() { + try { + this.task.run(); + } + catch (Throwable t) { + if (errorHandler != null) { + errorHandler.handle(t); + } + else if (logger.isWarnEnabled()) { + logger.warn("error occurred in task but no 'errorHandler' is available", t); + } + } + if (this.shouldRepeat && isRunning()) { + TaskRunner runner = new TaskRunner(this.task); + runner.setShouldRepeat(true); + scheduleServiceProvider.execute(runner); + } + } + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/spi/ScheduleServiceProvider.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/spi/ScheduleServiceProvider.java new file mode 100644 index 0000000000..88586f81e4 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/spi/ScheduleServiceProvider.java @@ -0,0 +1,43 @@ +/* + * 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. + * 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.scheduling.spi; + +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + +/** + * An SPI interface providing an abstraction of thread pooling and scheduling operations, allowing to switch between + * different platforms (such as Java 5's java.util.concurrent, Quartz, etc) at runtime. + * + * @author Marius Bogoevici + */ +public interface ScheduleServiceProvider { + + void execute(Runnable runnable); + + void shutdown(boolean waitForTasksToCompleteOnShutdown); + + ScheduledFuture scheduleWithInitialDelay(Runnable runnable, long initialDelay, TimeUnit timeUnit) + throws UnschedulableTaskException; + + ScheduledFuture scheduleAtFixedRate(Runnable runnable, long initialDelay, long period, TimeUnit timeUnit) + throws UnschedulableTaskException; + + ScheduledFuture scheduleWithFixedDelay(Runnable runnable, long initialDelay, long delay, TimeUnit timeUnit) + throws UnschedulableTaskException; + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/spi/SimpleScheduleServiceProvider.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/spi/SimpleScheduleServiceProvider.java new file mode 100644 index 0000000000..25f982f837 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/spi/SimpleScheduleServiceProvider.java @@ -0,0 +1,65 @@ +/* + * 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. + * 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.scheduling.spi; + +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + +/** + * An implementation + * @author Marius Bogoevici + */ +public class SimpleScheduleServiceProvider implements ScheduleServiceProvider { + + private final ScheduledExecutorService executor; + + public SimpleScheduleServiceProvider(ScheduledExecutorService executor) { + this.executor = executor; + } + + public void execute(Runnable runnable) { + this.executor.execute(runnable); + } + + public void shutdown(boolean waitForTasksToCompleteOnShutdown) { + if (this.executor.isShutdown()) { + return; + } + if (waitForTasksToCompleteOnShutdown) { + this.executor.shutdown(); + } + else { + this.executor.shutdownNow(); + } + } + + public ScheduledFuture scheduleWithInitialDelay(Runnable runnable, long initialDelay, TimeUnit timeUnit) { + return this.executor.schedule(runnable, initialDelay, timeUnit); + } + + public ScheduledFuture scheduleAtFixedRate(Runnable runnable, long initialDelay, long period, + TimeUnit timeUnit) { + return this.executor.scheduleAtFixedRate(runnable, initialDelay, period, timeUnit); + } + + public ScheduledFuture scheduleWithFixedDelay(Runnable runnable, long initialDelay, long delay, + TimeUnit timeUnit) { + return this.executor.scheduleWithFixedDelay(runnable, initialDelay, delay, timeUnit); + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/spi/UnschedulableTaskException.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/spi/UnschedulableTaskException.java new file mode 100644 index 0000000000..10e460cf4d --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/spi/UnschedulableTaskException.java @@ -0,0 +1,46 @@ +/* + * 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. + * 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.scheduling.spi; + +import org.springframework.integration.scheduling.Schedule; + +/** + * An exception indicating that a {@link org.springframework.integration.scheduling.SchedulableTask} + * could not be scheduled. The typical reason is that the {@link ScheduleServiceProvider} does not + * understand the {@link org.springframework.integration.scheduling.Schedule}. + * + * @author Marius Bogoevici + */ +public class UnschedulableTaskException extends RuntimeException { + + private Schedule schedule; + + public UnschedulableTaskException(Schedule schedule) { + super(); + this.schedule = schedule; + } + + public UnschedulableTaskException(String message, Schedule schedule) { + super(message); + this.schedule = schedule; + } + + public Schedule getSchedule() { + return schedule; + } + +} 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 3fc7bbeddb..68f0353630 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 @@ -42,7 +42,7 @@ import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.channel.config.ChannelParserTests; import org.springframework.integration.dispatcher.DirectChannel; import org.springframework.integration.handler.TestHandlers; -import org.springframework.integration.scheduling.SimpleTaskScheduler; +import org.springframework.integration.scheduling.spi.ProviderTaskScheduler; /** * @author Mark Fisher @@ -177,7 +177,7 @@ public class MessageBusParserTests { context.getBean(AbstractApplicationContext.APPLICATION_EVENT_MULTICASTER_BEAN_NAME); DirectFieldAccessor accessor = new DirectFieldAccessor(multicaster); Object taskExecutor = accessor.getPropertyValue("taskExecutor"); - assertEquals(SimpleTaskScheduler.class, taskExecutor.getClass()); + assertEquals(ProviderTaskScheduler.class, taskExecutor.getClass()); } @Test