diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/AbstractMessagingTaskScheduler.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/AbstractMessagingTaskScheduler.java
new file mode 100644
index 0000000000..4dc81cb3e1
--- /dev/null
+++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/AbstractMessagingTaskScheduler.java
@@ -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);
+
+}
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/DefaultMessagingTask.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/DefaultMessagingTask.java
new file mode 100644
index 0000000000..e3a8ab29bd
--- /dev/null
+++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/DefaultMessagingTask.java
@@ -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;
+		}
+	}
+
+}
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTask.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTask.java
new file mode 100644
index 0000000000..6a132571bc
--- /dev/null
+++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTask.java
@@ -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();
+
+}
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTaskScheduler.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTaskScheduler.java
new file mode 100644
index 0000000000..8d3446bb5a
--- /dev/null
+++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTaskScheduler.java
@@ -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);
+
+}
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollingSchedule.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollingSchedule.java
new file mode 100644
index 0000000000..882764abd5
--- /dev/null
+++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollingSchedule.java
@@ -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;
+	}
+
+}
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/Schedule.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/Schedule.java
new file mode 100644
index 0000000000..bbde1a13b6
--- /dev/null
+++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/Schedule.java
@@ -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 {
+
+}
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java
new file mode 100644
index 0000000000..5975092234
--- /dev/null
+++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java
@@ -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));
+		}
+	}
+
+}