Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

PollingSourceEndpoint delegates to PollingDispatcher. Added sendTimeout to SimpleDispatcher.

Browse Source
This commit is contained in:
Mark Fisher
2008-04-24 14:25:34 +00:00
parent 9c67c1516f
commit 467e0f803d
6 changed files with 88 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java
View File

@@ -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 '<em>maxMessagesPerTask</em>' property of the provided
* {@link DispatcherPolicy}, and the timeout for each receive call is
* determined by the policy's '<em>receiveTimeout</em>' 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<Message<?>> poll() {
List<Message<?>> messages = new LinkedList<Message<?>>();
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<Message<?>> messages = this.poller.poll();
List<Message<?>> 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"));
}
}
}
}

14
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java
View File

@@ -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<Target> targets = new CopyOnWriteArrayList<Target>();
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;
}

90
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SourcePoller.java
View File

@@ -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 '<em>maxMessagesPerTask</em>'
* property, and the timeout for each receive call is determined by the '<em>timeout</em>'
* 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<Message<?>> poll() {
List<Message<?>> messages = new LinkedList<Message<?>>();
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;
}
}

74
spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java
View File

@@ -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<Message<?>> poll(int limit) {
List<Message<?>> results = new ArrayList<Message<?>>();
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<Message<?>> 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();
}
}

2
spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java
View File

@@ -51,7 +51,7 @@ public class EndpointParserTests {
TestHandler handler = (TestHandler) context.getBean("testHandler");
assertNull(handler.getMessageString());
channel.send(new GenericMessage<String>(1, "test"));
handler.getLatch().await(50, TimeUnit.MILLISECONDS);
handler.getLatch().await(500, TimeUnit.MILLISECONDS);
assertEquals("test", handler.getMessageString());
}

28
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SourcePollerTests.java → spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingDispatcherTests.java
View File

@@ -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<Message<?>> results = poller.poll();
PollingDispatcher dispatcher = new PollingDispatcher(channel, null);
Collection<Message<?>> 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<Message<?>> results = poller.poll();
PollingDispatcher dispatcher = new PollingDispatcher(channel, null);
Collection<Message<?>> 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<Message<?>> 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<Message<?>> results = poller.poll();
PollingDispatcher dispatcher = new PollingDispatcher(channel, null);
Collection<Message<?>> 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<Message<?>> iter = results.iterator();
assertEquals("test2", iter.next().getPayload());