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

Added messaging task scheduler abstraction.

Browse Source
This commit is contained in:
Mark Fisher
2008-01-13 18:44:30 +00:00
parent e1b1b925ed
commit 9737b329dc
7 changed files with 375 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
spring-integration-core/src/main/java/org/springframework/integration/scheduling/AbstractMessagingTaskScheduler.java Normal file
View File

@@ -0,0 +1,41 @@
/*
* Copyright 2002-2007 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 MessagingTaskScheduler} implementations.
*
* @author Mark Fisher
*/
public abstract class AbstractMessagingTaskScheduler implements MessagingTaskScheduler {
public boolean prefersShortLivedTasks() {
return true;
}
/**
* Submit a task to be run once.
*/
public void execute(Runnable task) {
this.schedule(new DefaultMessagingTask(task));
}
public abstract ScheduledFuture<?> schedule(MessagingTask task);
}

61
spring-integration-core/src/main/java/org/springframework/integration/scheduling/DefaultMessagingTask.java Normal file
View File

@@ -0,0 +1,61 @@
/*
* Copyright 2002-2007 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 wrapper for {@link Runnable Runnables} that provides a schedule and also
* captures any error that may be thrown from the run() method.
*
* @author Mark Fisher
*/
public class DefaultMessagingTask implements MessagingTask {
private Runnable runnable;
private Schedule schedule;
private Throwable lastError;
public DefaultMessagingTask(Runnable runnable) {
this(runnable, null);
}
public DefaultMessagingTask(Runnable runnable, Schedule schedule) {
this.runnable = runnable;
this.schedule = schedule;
}
public Schedule getSchedule() {
return this.schedule;
}
public Throwable getLastError() {
return this.lastError;
}
public void run() {
try {
this.runnable.run();
}
catch (Throwable t) {
this.lastError = t;
}
}
}

33
spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTask.java Normal file
View File

@@ -0,0 +1,33 @@
/*
* Copyright 2002-2007 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;
/**
* An extension of runnable that provides metadata for a
* {@link MessagingTaskScheduler}.
*
* @author Mark Fisher
*/
public interface MessagingTask extends Runnable {
Schedule getSchedule();
Throwable getLastError();
void run();
}

32
spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTaskScheduler.java Normal file
View File

@@ -0,0 +1,32 @@
/*
* Copyright 2002-2007 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.scheduling.SchedulingTaskExecutor;
/**
* Base interface for {@link MessagingTask} schedulers.
*
* @author Mark Fisher
*/
public interface MessagingTaskScheduler extends SchedulingTaskExecutor {
ScheduledFuture<?> schedule(MessagingTask task);
}

86
spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollingSchedule.java Normal file
View File

@@ -0,0 +1,86 @@
/*
* Copyright 2002-2007 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.TimeUnit;
import org.springframework.util.Assert;
/**
* Scheduling metadata for a polling task.
*
* @author Mark Fisher
*/
public class PollingSchedule implements Schedule {
private long period;
private long initialDelay = 0;
private TimeUnit timeUnit = TimeUnit.MILLISECONDS;
private boolean fixedRate = false;
/**
* Create a fixed-delay schedule with no initial delay.
*
* @param period the polling period in milliseconds
*/
public PollingSchedule(long period) {
this.period = period;
}
public long getPeriod() {
return this.period;
}
/**
* Set the polling period. A period of 0 indicates a repeating task.
*/
public void setPeriod(long period) {
this.period = period;
}
public long getInitialDelay() {
return this.initialDelay;
}
public void setInitialDelay(long initialDelay) {
Assert.isTrue(initialDelay >= 0, "'initialDelay' must not be negative");
this.initialDelay = initialDelay;
}
public TimeUnit getTimeUnit() {
return this.timeUnit;
}
public void setTimeUnit(TimeUnit timeUnit) {
Assert.notNull(timeUnit, "'timeUnit' must not be null");
this.timeUnit = timeUnit;
}
public boolean getFixedRate() {
return this.fixedRate;
}
public void setFixedRate(boolean fixedRate) {
this.fixedRate = fixedRate;
}
}

28
spring-integration-core/src/main/java/org/springframework/integration/scheduling/Schedule.java Normal file
View File

@@ -0,0 +1,28 @@
/*
* Copyright 2002-2007 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 marker interface for scheduling metadata. Implementations of this interface
* will provide the information necessary for a {@link MessagingTaskScheduler}
* implementation to schedule tasks.
*
* @author Mark Fisher
*/
public interface Schedule {
}

94
spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java Normal file
View File

@@ -0,0 +1,94 @@
/*
* Copyright 2002-2007 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.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
/**
* An implementation of {@link MessagingTaskScheduler} that understands
* {@link PollingSchedule PollingSchedules}.
*
* @author Mark Fisher
*/
public class SimpleMessagingTaskScheduler extends AbstractMessagingTaskScheduler implements InitializingBean {
private ScheduledExecutorService executor;
private int corePoolSize = 10;
public void setExecutor(ScheduledExecutorService executor) {
Assert.notNull(executor, "'executor' must not be null");
this.executor = executor;
}
public void setCorePoolSize(int corePoolSize) {
Assert.isTrue(corePoolSize > 0, "'corePoolSize' must be greater than 0");
this.corePoolSize = corePoolSize;
}
public void afterPropertiesSet() {
if (this.executor == null) {
this.executor = new ScheduledThreadPoolExecutor(this.corePoolSize);
}
}
@Override
public ScheduledFuture<?> schedule(MessagingTask task) {
if (this.executor == null) {
this.afterPropertiesSet();
}
Schedule schedule = task.getSchedule();
if (schedule == null) {
return this.executor.schedule(task, 0, TimeUnit.MILLISECONDS);
}
if (schedule instanceof PollingSchedule) {
PollingSchedule ps = (PollingSchedule) schedule;
if (ps.getPeriod() <= 0) {
return this.executor.schedule(new RepeatingTask(task), ps.getInitialDelay(), ps.getTimeUnit());
}
if (ps.getFixedRate()) {
return this.executor.scheduleAtFixedRate(task, ps.getInitialDelay(), ps.getPeriod(), ps.getTimeUnit());
}
return this.executor.scheduleWithFixedDelay(task, ps.getInitialDelay(), ps.getPeriod(), ps.getTimeUnit());
}
throw new UnsupportedOperationException(this.getClass().getName() + " does not support schedule type '"
+ schedule.getClass().getName() + "'");
}
private class RepeatingTask implements Runnable {
private MessagingTask task;
RepeatingTask(MessagingTask task) {
this.task = task;
}
public void run() {
task.run();
executor.execute(new RepeatingTask(task));
}
}
}