diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSourceAdapter.java
index 4232a5b74d..e0c7fd8f1a 100644
--- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSourceAdapter.java
+++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSourceAdapter.java
@@ -20,10 +20,10 @@ import java.io.File;
 import java.io.FileFilter;
 import java.io.FilenameFilter;
 
-import org.springframework.integration.adapter.PollableSource;
 import org.springframework.integration.adapter.PollingSourceAdapter;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessagingException;
+import org.springframework.integration.message.PollableSource;
 import org.springframework.util.Assert;
 
 /**
@@ -84,6 +84,7 @@ public class FileSourceAdapter extends PollingSourceAdapter<Object> implements P
 		if (this.fileNameGenerator != null) {
 			this.mapper.setFileNameGenerator(this.fileNameGenerator);
 		}
+		super.initialize();
 	}
 
 	public Message<Object> poll() {
diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSourceAdapter.java
index af1c9efdfe..795947fe3c 100644
--- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSourceAdapter.java
+++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSourceAdapter.java
@@ -28,7 +28,6 @@ import org.apache.commons.net.ftp.FTP;
 import org.apache.commons.net.ftp.FTPClient;
 import org.apache.commons.net.ftp.FTPFile;
 
-import org.springframework.integration.adapter.PollableSource;
 import org.springframework.integration.adapter.PollingSourceAdapter;
 import org.springframework.integration.adapter.file.ByteArrayFileMapper;
 import org.springframework.integration.adapter.file.FileNameGenerator;
@@ -36,6 +35,7 @@ import org.springframework.integration.adapter.file.TextFileMapper;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageMapper;
 import org.springframework.integration.message.MessagingException;
+import org.springframework.integration.message.PollableSource;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -120,6 +120,7 @@ public class FtpSourceAdapter extends PollingSourceAdapter<Object> implements Po
 		else {
 			this.mapper = new ByteArrayFileMapper(this.localWorkingDirectory);
 		}
+		super.initialize();
 	}
 
 	@Override
diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java
index 7edc2ba92d..1589a6c674 100644
--- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java
+++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java
@@ -19,9 +19,9 @@ package org.springframework.integration.adapter.jms;
 import javax.jms.ConnectionFactory;
 import javax.jms.Destination;
 
-import org.springframework.integration.adapter.PollableSource;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.PollableSource;
 import org.springframework.jms.core.JmsTemplate;
 
 /**
diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java
index ad8d5403a5..182ef2f638 100644
--- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java
+++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java
@@ -20,10 +20,10 @@ import java.io.BufferedInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 
-import org.springframework.integration.adapter.PollableSource;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessagingException;
+import org.springframework.integration.message.PollableSource;
 
 /**
  * A pollable source for receiving bytes from an {@link InputStream}.
diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java
index 3c51415c13..8457ce51e6 100644
--- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java
+++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java
@@ -20,8 +20,8 @@ import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.Reader;
 
-import org.springframework.integration.adapter.PollableSource;
 import org.springframework.integration.message.MessagingException;
+import org.springframework.integration.message.PollableSource;
 import org.springframework.integration.message.StringMessage;
 import org.springframework.util.Assert;
 
diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java
index b8809ba03f..e7e4855884 100644
--- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java
+++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java
@@ -38,6 +38,7 @@ public class CharacterStreamSourceAdapterTests {
 		MessageChannel channel = new SimpleChannel();
 		CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(reader);
 		adapter.setChannel(channel);
+		adapter.setInitialDelay(10000);
 		adapter.start();
 		int count = adapter.processMessages();
 		assertEquals(1, count);
@@ -55,6 +56,7 @@ public class CharacterStreamSourceAdapterTests {
 		StringReader reader = new StringReader("test");
 		MessageChannel channel = new SimpleChannel();
 		CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(reader);
+		adapter.setInitialDelay(10000);
 		adapter.setChannel(channel);
 		adapter.setMaxMessagesPerTask(5);
 		adapter.start();
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingSource.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingSource.java
index 7a1e70f91f..3352dbabbb 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingSource.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingSource.java
@@ -23,6 +23,7 @@ import org.springframework.integration.ConfigurationException;
 import org.springframework.integration.handler.HandlerMethodInvoker;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.PollableSource;
 import org.springframework.integration.util.MethodValidator;
 import org.springframework.util.Assert;
 
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java
index 6aba0a2ec5..a070bc7af1 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java
@@ -23,8 +23,10 @@ import java.util.concurrent.Executors;
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.ConfigurationException;
 import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.dispatcher.SynchronousChannel;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageMapper;
+import org.springframework.integration.message.PollableSource;
 import org.springframework.integration.scheduling.MessagingTask;
 import org.springframework.integration.scheduling.MessagingTaskScheduler;
 import org.springframework.integration.scheduling.MessagingTaskSchedulerAware;
@@ -50,10 +52,10 @@ public class PollingSourceAdapter<T> extends AbstractSourceAdapter<T> implements
 
 	private volatile int maxMessagesPerTask = 1;
 
-	private volatile boolean starting;
-
 	private volatile boolean running;
 
+	private final Object lifecycleMonitor = new Object();
+
 
 	/**
 	 * Create a new adapter for the given source.
@@ -106,28 +108,38 @@ public class PollingSourceAdapter<T> extends AbstractSourceAdapter<T> implements
 		if (this.source == null) {
 			throw new ConfigurationException("source must not be null");
 		}
+		if (this.getChannel() instanceof SynchronousChannel) {
+			((SynchronousChannel) this.getChannel()).setSource(this.source);
+		}
 	}
 
 	public void start() {
-		if (this.isRunning() || this.starting) {
-			return;
-		}
-		this.starting = true;
-		if (!this.isInitialized()) {
-			this.afterPropertiesSet();
-		}
-		if (this.scheduler == null) {
-			if (logger.isInfoEnabled()) {
-				logger.info("no task scheduler has been provided, will create one");
+		synchronized (this.lifecycleMonitor) {
+			if (this.isRunning()) {
+				return;
 			}
-			this.scheduler = new SimpleMessagingTaskScheduler(Executors.newSingleThreadScheduledExecutor());
+			if (!this.isInitialized()) {
+				this.afterPropertiesSet();
+			}
+			if (this.getChannel() instanceof SynchronousChannel) {
+				if (logger.isInfoEnabled()) {
+					logger.info("source adapter configured on synchronous channel, not scheduling");
+				}
+				this.running = true;
+				return;
+			}
+			if (this.scheduler == null) {
+				if (logger.isInfoEnabled()) {
+					logger.info("no task scheduler has been provided, will create one");
+				}
+				this.scheduler = new SimpleMessagingTaskScheduler(Executors.newSingleThreadScheduledExecutor());
+			}
+			this.running = true;
 		}
 		if (!this.scheduler.isRunning()) {
 			this.scheduler.start();
 		}
 		this.scheduler.schedule(new PollingSourceAdapterTask());
-		this.running = true;
-		this.starting = false;
 	}
 
 	public void stop() {
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java
index c8de742864..62d9c5906f 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java
@@ -22,11 +22,11 @@ import java.util.Queue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import org.springframework.integration.adapter.PollableSource;
 import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.integration.channel.DispatcherPolicy;
 import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.PollableSource;
 import org.springframework.integration.message.selector.MessageSelector;
 
 /**
@@ -48,7 +48,7 @@ public class SynchronousChannel extends AbstractMessageChannel {
 	private static final ThreadLocalMessageHolder messageHolder = new ThreadLocalMessageHolder();
 
 
-	private final PollableSource<?> source;
+	private volatile PollableSource<?> source;
 
 	private final MessageDistributor distributor;
 
@@ -66,6 +66,10 @@ public class SynchronousChannel extends AbstractMessageChannel {
 	}
 
 
+	public void setSource(PollableSource<?> source) {
+		this.source = source;
+	}
+
 	public void addHandler(MessageHandler handler) {
 		this.distributor.addHandler(handler);
 		this.handlerCount.incrementAndGet();
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollableSource.java b/spring-integration-core/src/main/java/org/springframework/integration/message/PollableSource.java
similarity index 88%
rename from spring-integration-core/src/main/java/org/springframework/integration/adapter/PollableSource.java
rename to spring-integration-core/src/main/java/org/springframework/integration/message/PollableSource.java
index 520b0c37d0..748f6842c7 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollableSource.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/message/PollableSource.java
@@ -14,9 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.integration.adapter;
-
-import org.springframework.integration.message.Message;
+package org.springframework.integration.message;
 
 /**
  * Interface for any external message source that can be polled.
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java
index 96dcac2d8a..eda141e952 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java
@@ -27,6 +27,7 @@ import org.junit.Test;
 import org.springframework.integration.channel.SimpleChannel;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.PollableSource;
 
 /**
  * @author Mark Fisher
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java
index ff79e0a50f..0b90c1085e 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java
@@ -29,7 +29,6 @@ import org.junit.Test;
 import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.integration.ConfigurationException;
-import org.springframework.integration.adapter.PollableSource;
 import org.springframework.integration.adapter.PollingSourceAdapter;
 import org.springframework.integration.adapter.SourceAdapter;
 import org.springframework.integration.channel.DispatcherPolicy;
@@ -41,6 +40,7 @@ import org.springframework.integration.message.ErrorMessage;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageDeliveryException;
+import org.springframework.integration.message.PollableSource;
 import org.springframework.integration.message.StringMessage;
 import org.springframework.integration.scheduling.Subscription;
 
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java
index 3d5519e2bf..25d119b0f7 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java
@@ -28,9 +28,9 @@ import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
 
-import org.springframework.integration.adapter.PollableSource;
 import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.PollableSource;
 import org.springframework.integration.message.StringMessage;
 
 /**