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 83af4f6738..a3f1b77d02 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 @@ -36,21 +36,36 @@ import org.springframework.util.Assert; */ public class PollingSourceAdapter implements SourceAdapter, MessageDispatcher { + private static int DEFAULT_PERIOD = 1000; + private PollableSource source; private MessageChannel channel; private MessageMapper mapper = new SimplePayloadMessageMapper(); - private ConsumerPolicy policy; + private ConsumerPolicy policy = ConsumerPolicy.newPollingPolicy(DEFAULT_PERIOD); private long sendTimeout = -1; - public PollingSourceAdapter(PollableSource source, MessageChannel channel, int pollInterval) { + public PollingSourceAdapter(PollableSource 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 mapper) { @@ -62,25 +77,15 @@ public class PollingSourceAdapter 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(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java index f968ceb264..8d1ec65df7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java +++ b/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(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/file/FileSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/file/FileSourceAdapter.java index b0f0bc8562..f442c8611b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/file/FileSourceAdapter.java +++ b/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 { - 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)); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java index ed91ac25d4..3b8e782fc1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java +++ b/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; }