Added property setters for PollingSourceAdapter and convenience factory method to ConsumerPolicy.

spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java

@@ -36,21 +36,36 @@ import org.springframework.util.Assert;
  */
 public class PollingSourceAdapter<T> implements SourceAdapter, MessageDispatcher {
 
+	private static int DEFAULT_PERIOD = 1000;
+
 	private PollableSource<T> source;
 
 	private MessageChannel channel;
 
 	private MessageMapper<?,T> mapper = new SimplePayloadMessageMapper<T>();
 
-	private ConsumerPolicy policy;
+	private ConsumerPolicy policy = ConsumerPolicy.newPollingPolicy(DEFAULT_PERIOD);
 
 	private long sendTimeout = -1;
 
 
-	public PollingSourceAdapter(PollableSource<T> source, MessageChannel channel, int pollInterval) {
+	public PollingSourceAdapter(PollableSource<T> source) {
+		Assert.notNull(source, "'source' must not be null");
 		this.source = source;
+	}
+
+	public void setChannel(MessageChannel channel) {
+		Assert.notNull(channel, "'channel' must not be null");
 		this.channel = channel;
-		this.initConsumerPolicy(pollInterval);
+	}
+
+	public void setPeriod(int period) {
+		Assert.isTrue(period > 0, "'period' must be a positive value");
+		this.policy.setPeriod(period);
+	}
+
+	public void setSendTimeout(long sendTimeout) {
+		this.sendTimeout = sendTimeout;
 	}
 
 	public void setMessageMapper(MessageMapper<?,T> mapper) {
@@ -62,25 +77,15 @@ public class PollingSourceAdapter<T> implements SourceAdapter, MessageDispatcher
 		return this.mapper;
 	}
 
-	public void setLimit(int limit) {
-		Assert.isTrue(limit > 0, "'limit' must be a positive value");
-		this.policy.setMaxMessagesPerTask(limit);
-	}
-
-	public void setSendTimeout(long sendTimeout) {
-		this.sendTimeout = sendTimeout;
+	public void setMaxMessagesPerTask(int maxMessagesPerTask) {
+		Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagesPerTask' must be a positive value");
+		this.policy.setMaxMessagesPerTask(maxMessagesPerTask);
 	}
 
 	public ConsumerPolicy getConsumerPolicy() {
 		return this.policy;
 	}
 
-	private void initConsumerPolicy(int pollInterval) {
-		ConsumerPolicy policy = new ConsumerPolicy();
-		policy.setPeriod(pollInterval);
-		this.policy = policy;
-	}
-
 	public int receiveAndDispatch() {
 		int messagesProcessed = 0;
 		int limit = this.policy.getMaxMessagesPerTask();

spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java

@@ -17,6 +17,7 @@
 package org.springframework.integration.adapter;
 
 import org.springframework.integration.bus.ConsumerPolicy;
+import org.springframework.integration.channel.MessageChannel;
 
 /**
  * Base interface for source adapters.
@@ -25,6 +26,8 @@ import org.springframework.integration.bus.ConsumerPolicy;
  */
 public interface SourceAdapter {
 
+	void setChannel(MessageChannel channel);
+
 	ConsumerPolicy getConsumerPolicy();
 
 }

spring-integration-core/src/main/java/org/springframework/integration/adapter/file/FileSourceAdapter.java

@@ -30,12 +30,14 @@ import org.springframework.util.Assert;
  */
 public class FileSourceAdapter extends PollingSourceAdapter<File> {
 
-	public FileSourceAdapter(File directory, MessageChannel channel, int pollInterval) {
-		this(directory, channel, pollInterval, true);
+	public FileSourceAdapter(File directory, MessageChannel channel, int period) {
+		this(directory, channel, period, true);
 	}
 
-	public FileSourceAdapter(File directory, MessageChannel channel, int pollInterval, boolean isTextBased) {
-		super(new FileSource(directory), channel, pollInterval);
+	public FileSourceAdapter(File directory, MessageChannel channel, int period, boolean isTextBased) {
+		super(new FileSource(directory));
+		this.setChannel(channel);
+		this.setPeriod(period);
 		if (isTextBased) {
 			this.setMessageMapper(new TextFileMapper(directory));
 		}

spring-integration-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java

@@ -18,6 +18,8 @@ package org.springframework.integration.bus;
 
 import java.util.concurrent.TimeUnit;
 
+import org.springframework.util.Assert;
+
 /**
  * A container for Message consumer configuration metadata.
  * 
@@ -59,6 +61,20 @@ public class ConsumerPolicy {
 	private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
 
 
+	/**
+	 * Factory method for a basic polling policy.
+	 * 
+	 *  @param period the polling interval
+	 */
+	public static ConsumerPolicy newPollingPolicy(int period) {
+		Assert.isTrue(period > 0, "'period' must be a positive value");
+		ConsumerPolicy policy = new ConsumerPolicy();
+		policy.setPeriod(period);
+		policy.setConcurrency(1);
+		policy.setMaxConcurrency(1);
+		return policy;
+	}
+
 	public int getInitialDelay() {
 		return this.initialDelay;
 	}