diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java index 9b48d9ed80..4a9894f2d2 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java @@ -25,9 +25,12 @@ import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; import org.springframework.beans.factory.xml.BeanDefinitionParserDelegate; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.core.Conventions; +import org.springframework.integration.ConfigurationException; +import org.springframework.integration.scheduling.CronSchedule; import org.springframework.integration.scheduling.PollingSchedule; +import org.springframework.integration.scheduling.Schedule; import org.springframework.util.StringUtils; -import org.springframework.util.xml.DomUtils; +import org.springframework.util.xml.DomUtils; /** * Shared utility methods for integration namespace parsers. @@ -141,17 +144,27 @@ public abstract class IntegrationNamespaceUtils { */ public static String parsePoller(String sourceBeanName, Element element, ParserContext parserContext) { BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(PollingDispatcherFactoryBean.class); - Long period = Long.valueOf(element.getAttribute("period")); - PollingSchedule schedule = new PollingSchedule(period); - String initialDelay = element.getAttribute("initial-delay"); - if (StringUtils.hasText(initialDelay)) { - schedule.setInitialDelay(Long.valueOf(initialDelay)); + Schedule schedule = null; + if (!(StringUtils.hasText(element.getAttribute("period")) ^ StringUtils.hasText(element.getAttribute("cron")))) { + throw new ConfigurationException("A element must define either a period " + + "or a cron expression (but not both)"); } - if ("true".equals(element.getAttribute("fixed-rate").toLowerCase())) { - schedule.setFixedRate(true); + if (StringUtils.hasText(element.getAttribute("period"))) { + Long period = Long.valueOf(element.getAttribute("period")); + schedule = new PollingSchedule(period); + String initialDelay = element.getAttribute("initial-delay"); + if (StringUtils.hasText(initialDelay)) { + ((PollingSchedule)schedule).setInitialDelay(Long.valueOf(initialDelay)); + } + if ("true".equals(element.getAttribute("fixed-rate").toLowerCase())) { + ((PollingSchedule)schedule).setFixedRate(true); + } + else { + ((PollingSchedule)schedule).setFixedRate(false); + } } - else { - schedule.setFixedRate(false); + if (StringUtils.hasText(element.getAttribute("cron"))) { + schedule = new CronSchedule(element.getAttribute("cron")); } IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor"); Element txElement = DomUtils.getChildElementByTagName(element, "transactional"); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd index 4efb75aa40..d996c79aea 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd @@ -227,7 +227,8 @@ - + + diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/CronSchedule.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/CronSchedule.java new file mode 100644 index 0000000000..f10b5bf740 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/CronSchedule.java @@ -0,0 +1,38 @@ +/* + * 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; + +/** + * A schedule using a Cron expression. + * + * @author Marius Bogoevici + */ +public class CronSchedule implements Schedule { + + private final String cronExpression; + + + public CronSchedule(String cronExpression) { + this.cronExpression = cronExpression; + } + + + public String getCronExpression() { + return this.cronExpression; + } + +} 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 index f3bea16607..888ec792c3 100644 --- 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 @@ -27,6 +27,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.DisposableBean; +import org.springframework.integration.scheduling.CronSchedule; import org.springframework.integration.scheduling.PollingSchedule; import org.springframework.integration.scheduling.SchedulableTask; import org.springframework.integration.scheduling.TaskScheduler; @@ -37,7 +38,7 @@ 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 */ @@ -147,33 +148,41 @@ public class ProviderTaskScheduler implements TaskScheduler, DisposableBean { 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()); + try { + TaskRunner runner = new TaskRunner(task); + ScheduledFuture future = null; + if (task.getSchedule() == null) { + future = this.scheduleServiceProvider.scheduleWithInitialDelay(runner, 0, TimeUnit.MILLISECONDS); } - else if (ps.getFixedRate()) { - future = this.scheduleServiceProvider.scheduleAtFixedRate(runner, ps.getInitialDelay(), ps.getPeriod(), ps.getTimeUnit()); + 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()); + } } - else { - future = this.scheduleServiceProvider.scheduleWithFixedDelay(runner, ps.getInitialDelay(), ps.getPeriod(), ps.getTimeUnit()); + else if (task.getSchedule() instanceof CronSchedule) { + future = this.scheduleServiceProvider.scheduleWithCronExpression(runner, + ((CronSchedule) task.getSchedule()).getCronExpression()); } + this.scheduledTasks.put(task, future); + if (logger.isDebugEnabled()) { + logger.debug("scheduled task: " + task); + } + return future; } - if (future == null) { - throw new UnsupportedOperationException(this.getClass().getName() + " does not support scheduleWithInitialDelay type '" - + task.getSchedule().getClass().getName() + "'"); + catch (Exception e) { + throw new UnschedulableTaskException(e, task.getSchedule()); } - this.scheduledTasks.put(task, future); - if (logger.isDebugEnabled()) { - logger.debug("scheduled task: " + task); - } - return future; } } 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 index 88586f81e4..8a97982c1b 100644 --- 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 @@ -32,12 +32,15 @@ public interface ScheduleServiceProvider { void shutdown(boolean waitForTasksToCompleteOnShutdown); ScheduledFuture scheduleWithInitialDelay(Runnable runnable, long initialDelay, TimeUnit timeUnit) - throws UnschedulableTaskException; + throws Exception; ScheduledFuture scheduleAtFixedRate(Runnable runnable, long initialDelay, long period, TimeUnit timeUnit) - throws UnschedulableTaskException; + throws Exception; ScheduledFuture scheduleWithFixedDelay(Runnable runnable, long initialDelay, long delay, TimeUnit timeUnit) - throws UnschedulableTaskException; + throws Exception; + + ScheduledFuture scheduleWithCronExpression(Runnable runnable, String cronExpression) + throws Exception; } 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 index 25f982f837..b02a6d80e6 100644 --- 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 @@ -62,4 +62,9 @@ public class SimpleScheduleServiceProvider implements ScheduleServiceProvider { return this.executor.scheduleWithFixedDelay(runnable, initialDelay, delay, timeUnit); } + public ScheduledFuture scheduleWithCronExpression(Runnable runnable, String cronExpression) + throws Exception { + throw new UnsupportedOperationException("Cron scheduling not supported"); + } + } 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 index 10e460cf4d..c92ae329d2 100644 --- 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 @@ -29,8 +29,8 @@ public class UnschedulableTaskException extends RuntimeException { private Schedule schedule; - public UnschedulableTaskException(Schedule schedule) { - super(); + public UnschedulableTaskException(Throwable t, Schedule schedule) { + super(t); this.schedule = schedule; }