diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java index 8b55f3509c..d718630009 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java @@ -16,11 +16,15 @@ package org.springframework.integration.dispatcher; +import java.util.LinkedList; import java.util.List; import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.BlockingSource; import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageDeliveryAware; +import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.PollableSource; import org.springframework.integration.scheduling.MessagingTask; import org.springframework.integration.scheduling.Schedule; @@ -28,13 +32,24 @@ import org.springframework.util.Assert; /** * A subclass of {@link SimpleDispatcher} that adds message retrieval - * capabilities and may be scheduled to run as a task. + * capabilities and may be scheduled to run as a task. It polls a source for + * {@link Message Messages}. The number of messages retrieved per poll is + * limited by the 'maxMessagesPerTask' property of the provided + * {@link DispatcherPolicy}, and the timeout for each receive call is + * determined by the policy's 'receiveTimeout' property. In + * general, it is recommended to use a value of 1 (the default) for + * 'maxMessagesPerTask' whenever a significant timeout is provided. Otherwise + * the poller may be holding on to available messages while waiting for + * additional messages. Note that the 'timeout' value is only relevant if the + * specified source is an implementation of {@link BlockingSource}. The default + * timeout value is 0 indicating that the method should return immediately + * rather than waiting for a {@link Message} to become available. * * @author Mark Fisher */ public class PollingDispatcher extends SimpleDispatcher implements MessagingTask { - private final SourcePoller poller; + private final PollableSource source; private final Schedule schedule; @@ -46,12 +61,9 @@ public class PollingDispatcher extends SimpleDispatcher implements MessagingTask public PollingDispatcher(PollableSource source, DispatcherPolicy dispatcherPolicy, Schedule schedule) { super(dispatcherPolicy); Assert.notNull(source, "source must not be null"); - this.poller = new SourcePoller(source); - if (!(source instanceof MessageChannel)) { - this.poller.setMaxMessagesPerTask(dispatcherPolicy.getMaxMessagesPerTask()); - this.poller.setTimeout(dispatcherPolicy.getReceiveTimeout()); - } + this.source = source; this.schedule = schedule; + this.dispatcherPolicy.setReceiveTimeout(0); } @@ -59,10 +71,38 @@ public class PollingDispatcher extends SimpleDispatcher implements MessagingTask return this.schedule; } + public List> poll() { + List> messages = new LinkedList>(); + int limit = this.dispatcherPolicy.getMaxMessagesPerTask(); + while (messages.size() < limit) { + Message message = null; + long timeout = this.dispatcherPolicy.getReceiveTimeout(); + if (this.source instanceof BlockingSource && timeout >= 0) { + message = ((BlockingSource) this.source).receive(timeout); + } + else { + message = this.source.receive(); + } + if (message == null) { + return messages; + } + messages.add(message); + } + return messages; + } + public void run() { - List> messages = this.poller.poll(); + List> messages = this.poll(); for (Message message : messages) { - this.dispatch(message); + boolean sent = this.dispatch(message); + if (this.source instanceof MessageDeliveryAware) { + if (sent) { + ((MessageDeliveryAware) this.source).onSend(message); + } + else { + ((MessageDeliveryAware) this.source).onFailure(new MessageDeliveryException(message, "failed to send message")); + } + } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java index 953c9342ff..233645776e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java @@ -27,6 +27,7 @@ import org.apache.commons.logging.LogFactory; import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.handler.MessageHandlerNotRunningException; import org.springframework.integration.handler.MessageHandlerRejectedExecutionException; +import org.springframework.integration.message.BlockingTarget; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.SubscribableSource; @@ -43,14 +44,20 @@ public class SimpleDispatcher implements MessageDispatcher, SubscribableSource { private final List targets = new CopyOnWriteArrayList(); - private final DispatcherPolicy dispatcherPolicy; + protected final DispatcherPolicy dispatcherPolicy; + + private volatile long sendTimeout; public SimpleDispatcher(DispatcherPolicy dispatcherPolicy) { - this.dispatcherPolicy = dispatcherPolicy; + this.dispatcherPolicy = (dispatcherPolicy != null) ? dispatcherPolicy : new DispatcherPolicy(); } + public void setSendTimeout(long sendTimeout) { + this.sendTimeout = sendTimeout; + } + public boolean subscribe(Target target) { return this.targets.add(target); } @@ -88,7 +95,8 @@ public class SimpleDispatcher implements MessageDispatcher, SubscribableSource { while (iter.hasNext()) { Target target = iter.next(); try { - boolean sent = target.send(message); + boolean sent = (target instanceof BlockingTarget && this.sendTimeout >= 0) ? + ((BlockingTarget) target).send(message, this.sendTimeout) : target.send(message); if (!this.dispatcherPolicy.isPublishSubscribe() && sent) { return true; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SourcePoller.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SourcePoller.java deleted file mode 100644 index 3f3f9d701e..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SourcePoller.java +++ /dev/null @@ -1,90 +0,0 @@ -/* - * Copyright 2002-2008 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.dispatcher; - -import java.util.LinkedList; -import java.util.List; - -import org.springframework.integration.channel.DispatcherPolicy; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.message.BlockingSource; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.PollableSource; -import org.springframework.util.Assert; - -/** - * Polls a source for {@link Message Messages}. The number of messages - * retrieved per poll is limited by the 'maxMessagesPerTask' - * property, and the timeout for each receive call is determined by the 'timeout' - * property. In general, it is recommended to use a value of 1 (the default) for - * 'maxMessagesPerTask' whenever a significant timeout is provided. Otherwise - * the poller may be holding on to available messages while waiting for - * additional messages. Note that the 'timeout' value is only relevant if the - * specified source is an implementation of {@link BlockingSource}. The default - * timeout value is 0 indicating that the method should return immediately - * rather than waiting for a {@link Message} to become available. - * - * @author Mark Fisher - */ -public class SourcePoller { - - private final PollableSource source; - - private volatile int maxMessagesPerTask = 1; - - private volatile long timeout = 0; - - - public SourcePoller(PollableSource source) { - Assert.notNull(source, "source must not be null"); - if (source instanceof MessageChannel) { - DispatcherPolicy dispatcherPolicy = ((MessageChannel) source).getDispatcherPolicy(); - this.setMaxMessagesPerTask(dispatcherPolicy.getMaxMessagesPerTask()); - this.setTimeout(dispatcherPolicy.getReceiveTimeout()); - } - this.source = source; - } - - - public void setMaxMessagesPerTask(int maxMessagesPerTask) { - Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagesPerTask' must be a positive value"); - this.maxMessagesPerTask = maxMessagesPerTask; - } - - public void setTimeout(long timeout) { - this.timeout = timeout; - } - - public List> poll() { - List> messages = new LinkedList>(); - while (messages.size() < this.maxMessagesPerTask) { - Message message = null; - if (this.source instanceof BlockingSource && this.timeout >= 0) { - message = ((BlockingSource) this.source).receive(this.timeout); - } - else { - message = this.source.receive(); - } - if (message == null) { - return messages; - } - messages.add(message); - } - return messages; - } - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java index 79c1690afc..179afd07d3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java @@ -16,13 +16,9 @@ package org.springframework.integration.endpoint; -import java.util.ArrayList; -import java.util.List; - +import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageDeliveryAware; -import org.springframework.integration.message.MessageDeliveryException; +import org.springframework.integration.dispatcher.PollingDispatcher; import org.springframework.integration.message.PollableSource; import org.springframework.integration.scheduling.MessagingTask; import org.springframework.integration.scheduling.PollingSchedule; @@ -37,77 +33,33 @@ import org.springframework.util.Assert; */ public class PollingSourceEndpoint extends AbstractSourceEndpoint implements MessagingTask { - private final PollingSchedule schedule; + private final DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); - private volatile long sendTimeout = 0; - - private volatile int maxMessagesPerTask = 1; + private final PollingDispatcher dispatcher; public PollingSourceEndpoint(PollableSource source, MessageChannel channel, PollingSchedule schedule) { super(source, channel); Assert.notNull(schedule, "schedule must not be null"); - this.schedule = schedule; + this.dispatcher = new PollingDispatcher(source, this.dispatcherPolicy, schedule); + this.dispatcher.subscribe(this.getChannel()); } - public void setSendTimeout(long sendTimeout) { - this.sendTimeout = sendTimeout; - } - public void setMaxMessagesPerTask(int maxMessagesPerTask) { - Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagesPerTask' must be at least one"); - this.maxMessagesPerTask = maxMessagesPerTask; + this.dispatcherPolicy.setMaxMessagesPerTask(maxMessagesPerTask); + } + + public void setSendTimeout(long sendTimeout) { + this.dispatcher.setSendTimeout(sendTimeout); } public Schedule getSchedule() { - return this.schedule; - } - - public List> poll(int limit) { - List> results = new ArrayList>(); - int count = 0; - while (count < limit) { - Message message = ((PollableSource) this.getSource()).receive(); - if (message == null) { - break; - } - results.add(message); - count++; - } - return results; - } - - protected boolean sendMessage(Message message) { - if (message == null) { - throw new IllegalArgumentException("message must not be null"); - } - boolean sent = (this.sendTimeout < 0) ? this.getChannel().send(message) : this.getChannel().send(message, this.sendTimeout); - if (this.getSource() instanceof MessageDeliveryAware) { - if (sent) { - ((MessageDeliveryAware) this.getSource()).onSend(message); - } - else { - ((MessageDeliveryAware) this.getSource()).onFailure(new MessageDeliveryException(message, "failed to send message")); - } - } - return sent; + return this.dispatcher.getSchedule(); } public void run() { - int messagesProcessed = 0; - List> messages = this.poll(this.maxMessagesPerTask); - for (Message message : messages) { - if (this.sendMessage(message)) { - messagesProcessed++; - } - else { - break; - } - } - if (logger.isDebugEnabled()) { - logger.debug("polling source task processed " + messagesProcessed + " messages"); - } + this.dispatcher.run(); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java index 12e8815a50..6496db0003 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java @@ -51,7 +51,7 @@ public class EndpointParserTests { TestHandler handler = (TestHandler) context.getBean("testHandler"); assertNull(handler.getMessageString()); channel.send(new GenericMessage(1, "test")); - handler.getLatch().await(50, TimeUnit.MILLISECONDS); + handler.getLatch().await(500, TimeUnit.MILLISECONDS); assertEquals("test", handler.getMessageString()); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SourcePollerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingDispatcherTests.java similarity index 83% rename from spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SourcePollerTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingDispatcherTests.java index 53e893d0c0..a92106f0fc 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SourcePollerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingDispatcherTests.java @@ -33,22 +33,22 @@ import org.springframework.integration.message.StringMessage; /** * @author Mark Fisher */ -public class SourcePollerTests { +public class PollingDispatcherTests { @Test public void testSingleMessagePerRetrieval() { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setReceiveTimeout(0); MessageChannel channel = new QueueChannel(5, dispatcherPolicy); - SourcePoller poller = new SourcePoller(channel); - Collection> results = poller.poll(); + PollingDispatcher dispatcher = new PollingDispatcher(channel, null); + Collection> results = dispatcher.poll(); assertTrue(results.isEmpty()); channel.send(new StringMessage("test1"), 0); channel.send(new StringMessage("test2"), 0); - results = poller.poll(); + results = dispatcher.poll(); assertEquals(1, results.size()); assertEquals("test1", results.iterator().next().getPayload()); - results = poller.poll(); + results = dispatcher.poll(); assertEquals(1, results.size()); assertEquals("test2", results.iterator().next().getPayload()); } @@ -59,18 +59,18 @@ public class SourcePollerTests { dispatcherPolicy.setReceiveTimeout(0); dispatcherPolicy.setMaxMessagesPerTask(2); MessageChannel channel = new QueueChannel(5, dispatcherPolicy); - SourcePoller poller = new SourcePoller(channel); - Collection> results = poller.poll(); + PollingDispatcher dispatcher = new PollingDispatcher(channel, null); + Collection> results = dispatcher.poll(); assertTrue(results.isEmpty()); channel.send(new StringMessage("test1"), 0); channel.send(new StringMessage("test2"), 0); channel.send(new StringMessage("test3"), 0); - results = poller.poll(); + results = dispatcher.poll(); assertEquals(2, results.size()); Iterator> iter = results.iterator(); assertEquals("test1", iter.next().getPayload()); assertEquals("test2", iter.next().getPayload()); - results = poller.poll(); + results = dispatcher.poll(); assertEquals(1, results.size()); assertEquals("test3", results.iterator().next().getPayload()); } @@ -81,17 +81,17 @@ public class SourcePollerTests { dispatcherPolicy.setReceiveTimeout(0); dispatcherPolicy.setMaxMessagesPerTask(1); MessageChannel channel = new QueueChannel(5, dispatcherPolicy); - SourcePoller poller = new SourcePoller(channel); - Collection> results = poller.poll(); + PollingDispatcher dispatcher = new PollingDispatcher(channel, null); + Collection> results = dispatcher.poll(); assertTrue(results.isEmpty()); channel.send(new StringMessage("test1"), 0); channel.send(new StringMessage("test2"), 0); channel.send(new StringMessage("test3"), 0); - results = poller.poll(); + results = dispatcher.poll(); assertEquals(1, results.size()); assertEquals("test1", results.iterator().next().getPayload()); - poller.setMaxMessagesPerTask(5); - results = poller.poll(); + dispatcherPolicy.setMaxMessagesPerTask(5); + results = dispatcher.poll(); assertEquals(2, results.size()); Iterator> iter = results.iterator(); assertEquals("test2", iter.next().getPayload());