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

Refactored all task scheduling and poller triggering to use functionality now included in the Spring 3.0 core, and removed Spring Integration specific code that is now handled by that corresponding code in the core.

Browse Source
This commit is contained in:
Mark Fisher
2009-08-04 01:25:53 +00:00
parent 7e6d4a6e27
commit 67b6ea048a
47 changed files with 399 additions and 1070 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierHandler.java
View File

@@ -23,7 +23,6 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -41,8 +40,7 @@ import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.scheduling.IntervalTrigger;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
/**
@@ -218,8 +216,8 @@ public abstract class AbstractMessageBarrierHandler<T extends Collection<? exten
return;
}
Assert.state(this.taskScheduler != null, "TaskScheduler must not be null");
this.reaperFutureTask = this.taskScheduler.schedule(new PrunerTask(),
new IntervalTrigger(this.reaperInterval, TimeUnit.MILLISECONDS));
this.reaperFutureTask = this.taskScheduler.scheduleWithFixedDelay(
new PrunerTask(), this.reaperInterval);
}
}

2
org.springframework.integration/src/main/java/org/springframework/integration/channel/ExecutorChannel.java
View File

@@ -22,8 +22,8 @@ import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.dispatcher.LoadBalancingStrategy;
import org.springframework.integration.dispatcher.UnicastingDispatcher;
import org.springframework.integration.util.ErrorHandler;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.scheduling.support.ErrorHandler;
import org.springframework.util.Assert;
/**

6
org.springframework.integration/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2008 the original author or authors.
* Copyright 2002-2009 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.
@@ -26,7 +26,7 @@ import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.util.ErrorHandler;
import org.springframework.scheduling.support.ErrorHandler;
import org.springframework.util.Assert;
/**
@@ -71,7 +71,7 @@ public class MessagePublishingErrorHandler implements ErrorHandler, BeanFactoryA
}
}
public final void handle(Throwable t) {
public final void handleError(Throwable t) {
Message<?> failedMessage = (t instanceof MessagingException) ?
((MessagingException) t).getFailedMessage() : null;
MessageChannel errorChannel = this.resolveErrorChannel(failedMessage);

2
org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java
View File

@@ -20,8 +20,8 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.util.ErrorHandler;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.scheduling.support.ErrorHandler;
/**
* A channel that sends Messages to each of its subscribers.

14
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ApplicationEventMulticasterParser.java
View File

@@ -23,12 +23,11 @@ import org.w3c.dom.Element;
import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.core.SpringVersion;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.util.StringUtils;
/**
@@ -57,8 +56,15 @@ public class ApplicationEventMulticasterParser extends AbstractSingleBeanDefinit
builder.addPropertyReference("taskExecutor", taskExecutorRef);
}
else {
TaskExecutor taskExecutor = IntegrationContextUtils.createThreadPoolTaskExecutor(1, 10, 0, "event-multicaster-");
builder.addPropertyValue("taskExecutor", taskExecutor);
BeanDefinitionBuilder executorBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor");
executorBuilder.addPropertyValue("corePoolSize", 1);
executorBuilder.addPropertyValue("maxPoolSize", 10);
executorBuilder.addPropertyValue("queueCapacity", 0);
executorBuilder.addPropertyValue("threadNamePrefix", "event-multicaster-");
String executorBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(
executorBuilder.getBeanDefinition(), parserContext.getRegistry());
builder.addPropertyReference("taskExecutor", executorBeanName);
}
String springVersion = SpringVersion.getVersion();
if (springVersion != null && springVersion.startsWith("2")) {

12
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/DefaultConfiguringBeanFactoryPostProcessor.java
View File

@@ -16,6 +16,8 @@
package org.springframework.integration.config.xml;
import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -27,7 +29,6 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.context.IntegrationContextUtils;
/**
@@ -111,12 +112,13 @@ class DefaultConfiguringBeanFactoryPostProcessor implements BeanFactoryPostProce
if (!registry.isBeanNameInUse(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)) {
if (logger.isInfoEnabled()) {
logger.info("No bean named '" + IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME +
"' has been explicitly defined. Therefore, a default SimpleTaskScheduler will be created.");
"' has been explicitly defined. Therefore, a default ThreadPoolTaskScheduler will be created.");
}
TaskExecutor taskExecutor = IntegrationContextUtils.createThreadPoolTaskExecutor(2, 100, 0, "task-scheduler-");
BeanDefinitionBuilder schedulerBuilder = BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".scheduling.SimpleTaskScheduler");
schedulerBuilder.addConstructorArgValue(taskExecutor);
"org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler");
schedulerBuilder.addPropertyValue("poolSize", 10);
schedulerBuilder.addPropertyValue("threadNamePrefix", "task-scheduler-");
schedulerBuilder.addPropertyValue("rejectedExecutionHandler", new CallerRunsPolicy());
BeanDefinitionBuilder errorHandlerBuilder = BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".channel.MessagePublishingErrorHandler");
errorHandlerBuilder.addPropertyReference("defaultErrorChannel", IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME);

20
org.springframework.integration/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java
View File

@@ -16,16 +16,11 @@
package org.springframework.integration.context;
import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* @author Mark Fisher
@@ -70,17 +65,4 @@ public abstract class IntegrationContextUtils {
return (T) bean;
}
public static ThreadPoolTaskExecutor createThreadPoolTaskExecutor(int coreSize, int maxSize, int queueCapacity, String threadPrefix) {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(coreSize);
executor.setMaxPoolSize(maxSize);
executor.setQueueCapacity(queueCapacity);
if (StringUtils.hasText(threadPrefix)) {
executor.setThreadFactory(new CustomizableThreadFactory(threadPrefix));
}
executor.setRejectedExecutionHandler(new CallerRunsPolicy());
executor.afterPropertiesSet();
return executor;
}
}

4
org.springframework.integration/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2008 the original author or authors.
* Copyright 2002-2009 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.
@@ -24,7 +24,7 @@ import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.integration.channel.BeanFactoryChannelResolver;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
/**

4
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2008 the original author or authors.
* Copyright 2002-2009 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.
@@ -22,7 +22,7 @@ import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.scheduling.TaskScheduler;
/**
* The base class for Message Endpoint implementations.

10
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2008 the original author or authors.
* Copyright 2002-2009 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.
@@ -28,9 +28,9 @@ import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.channel.BeanFactoryChannelResolver;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.scheduling.Trigger;
import org.springframework.integration.util.ErrorHandler;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.support.ErrorHandler;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
@@ -223,8 +223,8 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
private boolean innerPoll() {
TransactionTemplate txTemplate = getTransactionTemplate();
if (txTemplate != null) {
return (Boolean) txTemplate.execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
return txTemplate.execute(new TransactionCallback<Boolean>() {
public Boolean doInTransaction(TransactionStatus status) {
return doPoll();
}
});

1
org.springframework.integration/src/main/java/org/springframework/integration/handler/DelayHandler.java
View File

@@ -40,7 +40,6 @@ import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.util.Assert;
/**

328
org.springframework.integration/src/main/java/org/springframework/integration/scheduling/CronSequenceGenerator.java
View File

@@ -1,328 +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.BitSet;
import java.util.Calendar;
import java.util.Date;
import java.util.GregorianCalendar;
import org.springframework.util.StringUtils;
/**
* Date sequence generator for a <a
* href="http://www.manpagez.com/man/5/crontab/">Crontab pattern</a> allowing
* client to specify a pattern that the sequence matches. The pattern is a list
* of 6 single space separated fields representing (second, minute, hour, day,
* month, weekday). Month and weekday names can be given as the first three
* letters of the English names.<br/>
* <br/>
*
* Example patterns
* <ul>
* <li>"0 0 * * * *" = the top of every hour of every day.</li>
* <li>"*&#47;10 * * * * *" = every ten seconds.</li>
* <li>"0 0 8-10 * * *" = 8, 9 and 10 o'clock of every day.</li>
* <li>"0 0 8-10/30 * * *" = 8:00, 8:30, 9:00, 9:30 and 10 o'clock every day.</li>
* <li>"0 0 9-17 * * MON-FRI" = on the hour nine-to-five weekdays</li>
* <li>"0 0 0 25 12 ?" = every Christmas Day at midnight</li>
* </ul>
*
* @author Dave Syer
*/
class CronSequenceGenerator {
private final BitSet seconds = new BitSet(60);
private final BitSet minutes = new BitSet(60);
private final BitSet hours = new BitSet(24);
private final BitSet daysOfWeek = new BitSet(7);
private final BitSet daysOfMonth = new BitSet(31);
private final BitSet months = new BitSet(12);
private final String pattern;
/**
* Construct a {@link CronSequenceGenerator} from the pattern provided.
*
* @param pattern a space separated list of time fields
*
* @throws IllegalArgumentException if the pattern cannot be parsed
*/
public CronSequenceGenerator(String pattern) throws IllegalArgumentException {
this.pattern = pattern;
parse(pattern);
}
/**
* Get the next {@link Date} in the sequence matching the Cron pattern and
* after the value provided. The return value will have a whole number of
* seconds, and will be after the input value.
*
* @param date a seed value
* @return the next value matching the pattern
*/
public Date next(Date date) {
Calendar calendar = new GregorianCalendar();
calendar.setTime(date);
// Truncate to the next whole second
calendar.add(Calendar.SECOND, 1);
calendar.set(Calendar.MILLISECOND, 0);
int second = calendar.get(Calendar.SECOND);
second = findNext(seconds, second, 60, calendar, Calendar.SECOND);
int minute = calendar.get(Calendar.MINUTE);
minute = findNext(minutes, minute, 60, calendar, Calendar.MINUTE, Calendar.SECOND);
int hour = calendar.get(Calendar.HOUR_OF_DAY);
hour = findNext(hours, hour, 24, calendar, Calendar.HOUR_OF_DAY, Calendar.MINUTE, Calendar.SECOND);
int dayOfWeek = calendar.get(Calendar.DAY_OF_WEEK);
int dayOfMonth = calendar.get(Calendar.DAY_OF_MONTH);
dayOfMonth = findNextDay(calendar, daysOfMonth, dayOfMonth, daysOfWeek, dayOfWeek, 366);
int month = calendar.get(Calendar.MONTH);
month = findNext(months, month, 12, calendar, Calendar.MONTH, Calendar.DAY_OF_MONTH, Calendar.HOUR_OF_DAY,
Calendar.MINUTE, Calendar.SECOND);
return calendar.getTime();
}
/**
* @param calendar
* @return
*/
private int findNextDay(Calendar calendar, BitSet daysOfMonth, int dayOfMonth, BitSet daysOfWeek, int dayOfWeek,
int max) {
int count = 0;
// the DAY_OF_WEEK values in java.util.Calendar start with 1 (Sunday),
// but in the cron pattern, they start with 0, so we subtract 1 here
while ((!daysOfMonth.get(dayOfMonth) || !daysOfWeek.get(dayOfWeek-1)) && count++ < max) {
calendar.add(Calendar.DAY_OF_MONTH, 1);
dayOfMonth = calendar.get(Calendar.DAY_OF_MONTH);
dayOfWeek = calendar.get(Calendar.DAY_OF_WEEK);
reset(calendar, Calendar.HOUR_OF_DAY, Calendar.MINUTE, Calendar.SECOND);
}
if (count > max) {
throw new IllegalStateException("Overflow in day for expression=" + pattern);
}
return dayOfMonth;
}
/**
* Search the bits provided for the next set bit after the value provided,
* and reset the calendar.
*
* @param bits a {@link BitSet} representing the allowed values of the field
* @param value the current value of the field
* @param max the largest value that the field can have
* @param calendar the calendar to increment as we move through the bits
* @param field the field to increment in the calendar (@see
* {@link Calendar} for the static constants defining valid fields)
* @param lowerOrders the Calendar field ids that should be reset (i.e. the
* ones of lower significance than the field of interest)
*
* @return the value of the calendar field that is next in the sequence
*/
private int findNext(BitSet bits, int value, int max, Calendar calendar, int field, int... lowerOrders) {
int nextValue = bits.nextSetBit(value);
//roll over if needed
if (nextValue == -1) {
calendar.add(field, max - value);
nextValue = bits.nextSetBit(0);
}
if (nextValue != value) {
calendar.set(field, nextValue);
reset(calendar, lowerOrders);
}
return nextValue;
}
/**
* Reset the calendar setting all the fields provided to zero.
*
* @param calendar
* @param fields
*/
private void reset(Calendar calendar, int... fields) {
for (int field : fields) {
calendar.set(field, 0);
}
}
/**
* @param expression
*/
private void parse(String expression) throws IllegalArgumentException {
String[] fields = StringUtils.tokenizeToStringArray(expression, " ");
if (fields.length != 6) {
throw new IllegalArgumentException(String.format(""
+ "cron expression must consist of 6 fields (found %d in %s)", fields.length, expression));
}
setNumberHits(seconds, fields[0], 60);
setNumberHits(minutes, fields[1], 60);
setNumberHits(hours, fields[2], 24);
setDaysOfMonth(daysOfMonth, fields[3], 31);
setNumberHits(months, replaceOrdinals(fields[4], "JAN,FEB,MAR,APR,MAY,JUN,JUL,AUG,SEP,OCT,NOV,DEC"), 12);
setDays(daysOfWeek, replaceOrdinals(fields[5], "SUN,MON,TUE,WED,THU,FRI,SAT"), 8);
if (daysOfWeek.get(7)) {
// Sunday can be represented as 0 or 7
daysOfWeek.set(0);
daysOfWeek.clear(7);
}
}
/**
* Replace the values in the commaSeparatedList (case insensitive) with
* their index in the list.
*
* @param value
* @param commaSeparatedList
* @return a new string with the values from the list replaced
*/
private String replaceOrdinals(String value, String commaSeparatedList) {
String[] list = StringUtils.commaDelimitedListToStringArray(commaSeparatedList);
for (int i = 0; i < list.length; i++) {
String item = list[i].toUpperCase();
value = StringUtils.replace(value.toUpperCase(), item, "" + i);
}
return value;
}
/**
* @param bits
* @param field
* @param max
*/
private void setDaysOfMonth(BitSet bits, String field, int max) {
// Days of month start with 1 (in Cron and Calendar) so add one
setDays(bits, field, max+1);
// ... and remove it from the front
bits.clear(0);
}
/**
* @param bits
* @param field
* @param max
*/
private void setDays(BitSet bits, String field, int max) {
if (field.contains("?")) {
field = "*";
}
setNumberHits(bits, field, max);
}
/**
* @param bits
* @param value
* @param max
* @return
*/
private void setNumberHits(BitSet bits, String value, int max) {
String[] fields = StringUtils.delimitedListToStringArray(value, ",");
for (String field : fields) {
if (!field.contains("/")) {
// Not an incrementer so it must be a range (possibly empty)
int[] range = getRange(field, max);
bits.set(range[0], range[1] + 1);
}
else {
String[] split = StringUtils.delimitedListToStringArray(field, "/");
if (split.length > 2) {
throw new IllegalArgumentException("Incrementer has more than two fields: " + field);
}
int[] range = getRange(split[0], max);
if (!split[0].contains("-")) {
range[1] = max - 1;
}
int delta = Integer.valueOf(split[1]);
for (int i = range[0]; i <= range[1]; i += delta) {
bits.set(i);
}
}
}
}
/**
* @param field
* @return
*/
private int[] getRange(String field, int max) {
int[] result = new int[2];
if (field.contains("*")) {
result[0] = 0;
result[1] = max-1;
return result;
}
if (!field.contains("-")) {
result[0] = result[1] = Integer.valueOf(field);
}
else {
String[] split = StringUtils.delimitedListToStringArray(field, "-");
if (split.length > 2) {
throw new IllegalArgumentException("Range has more than two fields: " + field);
}
result[0] = Integer.valueOf(split[0]);
result[1] = Integer.valueOf(split[1]);
}
return result;
}
/**
* @see Object#equals(Object)
*/
@Override
public boolean equals(Object obj) {
if (!(obj instanceof CronSequenceGenerator)) {
return false;
}
CronSequenceGenerator cron = (CronSequenceGenerator) obj;
return cron.months.equals(months) && cron.daysOfMonth.equals(daysOfMonth) && cron.daysOfWeek.equals(daysOfWeek)
&& cron.hours.equals(hours) && cron.minutes.equals(minutes) && cron.seconds.equals(seconds);
}
/**
* @see Object#hashCode()
*/
@Override
public int hashCode() {
return 37 + 17 * months.hashCode() + 29 * daysOfMonth.hashCode() + 37 * daysOfWeek.hashCode() + 41
* hours.hashCode() + 53 * minutes.hashCode() + 61 * seconds.hashCode();
}
@Override
public String toString() {
return getClass().getSimpleName() + ": " + pattern;
}
}

65
org.springframework.integration/src/main/java/org/springframework/integration/scheduling/CronTrigger.java
View File

@@ -1,65 +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.Date;
/**
* A trigger that uses a cron expression. See {@link CronSequenceGenerator}
* for a detailed description of the expression pattern syntax.
*
* @author Mark Fisher
*/
public class CronTrigger implements Trigger {
private final CronSequenceGenerator cronSequenceGenerator;
/**
* Create a trigger for the given cron expression.
* See {@link CronSequenceGenerator}.
*/
public CronTrigger(String expression) throws IllegalArgumentException {
this.cronSequenceGenerator = new CronSequenceGenerator(expression);
}
/**
* Return the next time a task should run. Determined by consulting this
* trigger's cron expression compared with the lastCompleteTime. If the
* lastCompleteTime is <code>null</code>, the current time is used.
*/
public Date getNextRunTime(Date lastScheduledRunTime, Date lastCompleteTime) {
Date date = (lastCompleteTime != null) ? lastCompleteTime : new Date();
return this.cronSequenceGenerator.next(date);
}
@Override
public boolean equals(Object other) {
if (other == null || !(other instanceof CronTrigger)) {
return false;
}
return this.cronSequenceGenerator.equals(
((CronTrigger) other).cronSequenceGenerator);
}
@Override
public int hashCode() {
return this.cronSequenceGenerator.hashCode();
}
}

12
org.springframework.integration/src/main/java/org/springframework/integration/scheduling/IntervalTrigger.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2008 the original author or authors.
* Copyright 2002-2009 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,6 +19,8 @@ package org.springframework.integration.scheduling;
import java.util.Date;
import java.util.concurrent.TimeUnit;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.TriggerContext;
import org.springframework.util.Assert;
/**
@@ -78,14 +80,14 @@ public class IntervalTrigger implements Trigger {
/**
* Returns the next time a task should run.
*/
public Date getNextRunTime(Date lastScheduledRunTime, Date lastCompleteTime) {
if (lastScheduledRunTime == null) {
public Date nextExecutionTime(TriggerContext triggerContext) {
if (triggerContext.lastScheduledExecutionTime() == null) {
return new Date(System.currentTimeMillis() + this.initialDelay);
}
else if (this.fixedRate) {
return new Date(lastScheduledRunTime.getTime() + this.interval);
return new Date(triggerContext.lastScheduledExecutionTime().getTime() + this.interval);
}
return new Date(lastCompleteTime.getTime() + this.interval);
return new Date(triggerContext.lastCompletionTime().getTime() + this.interval);
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2008 the original author or authors.
* Copyright 2002-2009 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,8 +19,8 @@ package org.springframework.integration.scheduling;
import java.util.List;
import org.aopalliance.aop.Advice;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.Trigger;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;

320
org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SimpleTaskScheduler.java
View File

@@ -1,320 +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.Collections;
import java.util.Date;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.channel.BeanFactoryChannelResolver;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.util.ErrorHandler;
import org.springframework.scheduling.SchedulingException;
import org.springframework.util.Assert;
/**
* An implementation of {@link TaskScheduler} that delegates to any instance
* of {@link TaskExecutor}.
*
* <p>This class implements ApplicationListener and provides an {@link #autoStartup}
* property. If <code>true</code>, the scheduler will start automatically upon
* receiving the {@link ContextRefreshedEvent}. Otherwise, it will require an
* explicit invocation of its {@link #start()} method. The default value is
* <code>true</code>. To require explicit startup, provide a value of
* <code>false</code> to the {@link #setAutoStartup(boolean)} method.
*
* @author Mark Fisher
* @author Marius Bogoevici
*/
public class SimpleTaskScheduler implements TaskScheduler, BeanFactoryAware, ApplicationListener, DisposableBean {
private final Log logger = LogFactory.getLog(this.getClass());
private final TaskExecutor executor;
private volatile boolean autoStartup = true;
private volatile ErrorHandler errorHandler;
private volatile SchedulerTask schedulerTask = null;
private final DelayQueue<TriggeredTask<?>> scheduledTasks = new DelayQueue<TriggeredTask<?>>();
private final Set<TriggeredTask<?>> executingTasks = Collections.synchronizedSet(new TreeSet<TriggeredTask<?>>());
private volatile boolean running;
private final ReentrantLock lifecycleLock = new ReentrantLock();
public SimpleTaskScheduler(TaskExecutor executor) {
Assert.notNull(executor, "executor must not be null");
this.executor = executor;
}
public void setAutoStartup(boolean autoStartup) {
this.autoStartup = autoStartup;
}
public void setErrorHandler(ErrorHandler errorHandler) {
this.errorHandler = errorHandler;
}
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
if (this.errorHandler == null) {
this.errorHandler = new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(beanFactory));
}
}
public final ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
Assert.notNull(task, "task must not be null");
TriggeredTask<Void> triggeredTask = new TriggeredTask<Void>(task, trigger);
return this.schedule(triggeredTask, null, null);
}
private <V> ScheduledFuture<V> schedule(TriggeredTask<V> triggeredTask, Date lastScheduledRunTime, Date lastCompleteTime) {
Date nextRunTime = triggeredTask.trigger.getNextRunTime(lastScheduledRunTime, lastCompleteTime);
if (nextRunTime != null) {
triggeredTask.setScheduledTime(nextRunTime);
this.scheduledTasks.offer(triggeredTask);
}
return triggeredTask;
}
// Lifecycle implementation
public final boolean isRunning() {
this.lifecycleLock.lock();
try {
return this.running;
}
finally {
this.lifecycleLock.unlock();
}
}
public final void start() {
this.lifecycleLock.lock();
try {
if (!this.running) {
this.executor.execute(this.schedulerTask = new SchedulerTask());
this.running = true;
if (logger.isInfoEnabled()) {
logger.info("started " + this);
}
}
}
finally {
this.lifecycleLock.unlock();
}
}
public final void stop() {
this.lifecycleLock.lock();
try {
if (this.running) {
this.schedulerTask.deactivate();
Thread executingThread = this.schedulerTask.executingThread.get();
if (executingThread != null) {
executingThread.interrupt();
}
this.scheduledTasks.clear();
synchronized (this.executingTasks) {
for (TriggeredTask<?> task : this.executingTasks) {
task.cancel(true);
}
this.executingTasks.clear();
}
this.schedulerTask = null;
this.running = false;
if (logger.isInfoEnabled()) {
logger.info("stopped " + this);
}
}
}
finally {
this.lifecycleLock.unlock();
}
}
public final void onApplicationEvent(ApplicationEvent event) {
if (event instanceof ContextRefreshedEvent && this.autoStartup) {
this.start();
}
}
public void destroy() throws Exception {
this.stop();
if (this.executor instanceof DisposableBean) {
if (logger.isInfoEnabled()) {
logger.info("shutting down TaskExecutor");
}
((DisposableBean) this.executor).destroy();
}
}
public boolean prefersShortLivedTasks() {
return true;
}
public void execute(Runnable task) {
this.executor.execute(task);
}
private class SchedulerTask implements Runnable {
private final AtomicReference<Thread> executingThread = new AtomicReference<Thread>();
private volatile boolean active = true;
public void run() {
if (!this.executingThread.compareAndSet(null, Thread.currentThread())) {
throw new SchedulingException("The SchedulerTask is already running.");
}
while (this.active) {
try {
TriggeredTask<?> task = SimpleTaskScheduler.this.scheduledTasks.take();
//if this thread is not active anymore, clear
if (this.active) {
SimpleTaskScheduler.this.executor.execute(task);
}
else {
scheduledTasks.offer(task);
}
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
this.executingThread.set(null);
}
public void deactivate() {
this.active = false;
}
}
/**
* Wrapper class that enables rescheduling of a task based on a Trigger.
*/
private class TriggeredTask<V> extends FutureTask<V> implements Delayed, ScheduledFuture<V> {
private final Trigger trigger;
private volatile Date scheduledTime;
public TriggeredTask(Runnable task, Trigger trigger) {
super(new ErrorHandlingRunnableWrapper(task), null);
this.trigger = trigger;
}
public void setScheduledTime(Date scheduledTime) {
this.scheduledTime = scheduledTime;
}
public void run() {
SimpleTaskScheduler.this.executingTasks.add(this);
super.runAndReset();
SimpleTaskScheduler.this.executingTasks.remove(this);
if (SimpleTaskScheduler.this.isRunning() && !this.isCancelled()) {
SimpleTaskScheduler.this.schedule(this, this.scheduledTime, new Date());
}
}
public int compareTo(Delayed other) {
long thisDelay = this.getDelay(TimeUnit.MILLISECONDS);
long otherDelay = other.getDelay(TimeUnit.MILLISECONDS);
if (thisDelay < otherDelay) {
return -1;
}
if (thisDelay == otherDelay) {
return 0;
}
return 1;
}
public long getDelay(TimeUnit unit) {
long now = new Date().getTime();
long scheduled = (this.scheduledTime != null) ? this.scheduledTime.getTime() : now;
return (scheduled > now) ? unit.convert(scheduled - now, TimeUnit.MILLISECONDS) : 0;
}
public synchronized boolean cancel(boolean mayInterruptIfRunning) {
if (!this.isCancelled()) {
SimpleTaskScheduler.this.scheduledTasks.remove(this);
}
return super.cancel(mayInterruptIfRunning);
}
}
/**
* Wrapper that catches any Throwable thrown by a target task and
* delegates to the {@link ErrorHandler} if available. If no error handler
* has been configured, the error will be logged at error-level.
*/
private class ErrorHandlingRunnableWrapper implements Runnable {
private final Runnable target;
public ErrorHandlingRunnableWrapper(Runnable target) {
this.target = target;
}
public void run() {
try {
this.target.run();
}
catch (Throwable t) {
if (SimpleTaskScheduler.this.errorHandler != null) {
SimpleTaskScheduler.this.errorHandler.handle(t);
}
else if (logger.isErrorEnabled()) {
logger.error("Error occurred in task but no 'errorHandler' is available.", t);
}
}
}
}
}

40
org.springframework.integration/src/main/java/org/springframework/integration/scheduling/TaskScheduler.java
View File

@@ -1,40 +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;
import org.springframework.context.Lifecycle;
/**
* Base interface for scheduling tasks.
*
* @author Mark Fisher
* @author Marius Bogoevici
* @author Iwein Fuld
*/
public interface TaskScheduler extends Lifecycle {
/**
* Schedules a task for multiple executions according to a Trigger.
*
* @param task Task to be run multiple times
* @param trigger Trigger that determines at which times the task should be run
*/
ScheduledFuture<?> schedule(Runnable task, Trigger trigger);
}

40
org.springframework.integration/src/main/java/org/springframework/integration/scheduling/Trigger.java
View File

@@ -1,40 +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.Date;
/**
* A strategy for providing the next time a task should run.
*
* @author Mark Fisher
*/
public interface Trigger {
/**
* Returns the next time that a task should run or <code>null</code> if the
* task should not run again.
*
* @param lastScheduledRunTime last time the relevant task was scheduled to
* run, or <code>null</code> if it has never been scheduled
* @param lastCompleteTime last time the relevant task finished or
* <code>null</code> if it did not run to completion
* @return next time that a task should run
*/
public Date getNextRunTime(Date lastScheduledRunTime, Date lastCompleteTime);
}

28
org.springframework.integration/src/main/java/org/springframework/integration/util/ErrorHandler.java
View File

@@ -1,28 +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.util;
/**
* Strategy for handling a {@link Throwable}.
*
* @author Mark Fisher
*/
public interface ErrorHandler {
void handle(Throwable t);
}

5
org.springframework.integration/src/main/java/org/springframework/integration/util/ErrorHandlingTaskExecutor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2008 the original author or authors.
* Copyright 2002-2009 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,6 +17,7 @@
package org.springframework.integration.util;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.support.ErrorHandler;
import org.springframework.util.Assert;
/**
@@ -49,7 +50,7 @@ public class ErrorHandlingTaskExecutor implements TaskExecutor {
task.run();
}
catch (Throwable t) {
errorHandler.handle(t);
errorHandler.handleError(t);
}
}
});