Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Adding CronSchedule and parsing support for <poller cron="cronExpression"/>. Internal changes to the SPI.

Browse Source
This commit is contained in:
Marius Bogoevici
2008-08-19 21:26:36 +00:00
parent 8058ec7484
commit 38bda54d4c
7 changed files with 108 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java
View File

@@ -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 <poller> 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");

3
org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd
View File

@@ -227,7 +227,8 @@
<xsd:sequence>
<xsd:element name="transactional" type="transactionalType" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attribute name="period" type="xsd:long" use="required"/>
<xsd:attribute name="period" type="xsd:long"/>
<xsd:attribute name="cron" type="xsd:string"/>
<xsd:attribute name="initial-delay" type="xsd:long"/>
<xsd:attribute name="fixed-rate" type="xsd:boolean"/>
<xsd:attribute name="receive-timeout" type="xsd:long"/>

38
org.springframework.integration/src/main/java/org/springframework/integration/scheduling/CronSchedule.java Normal file
View File

@@ -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;
}
}

55
org.springframework.integration/src/main/java/org/springframework/integration/scheduling/spi/ProviderTaskScheduler.java
View File

@@ -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;
}
}

9
org.springframework.integration/src/main/java/org/springframework/integration/scheduling/spi/ScheduleServiceProvider.java
View File

@@ -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;
}

5
org.springframework.integration/src/main/java/org/springframework/integration/scheduling/spi/SimpleScheduleServiceProvider.java
View File

@@ -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");
}
}

4
org.springframework.integration/src/main/java/org/springframework/integration/scheduling/spi/UnschedulableTaskException.java
View File

@@ -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;
}