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

Removed 'maxMessagesPerTask' and 'receiveTimeout' properties from DispatcherPolicy and added those same properties to PollingDispatcherTask. The MessageDispatcher interface now extends the MessageTarget interface. Therefore, the 'dispatch(Message)' method has been replaced with 'send(Message)'.

Browse Source
This commit is contained in:
Mark Fisher
2008-07-02 22:56:00 +00:00
parent 7cefe22b96
commit d998095421
15 changed files with 58 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
org.springframework.integration/src/main/java/org/springframework/integration/channel/DispatcherPolicy.java
View File

@@ -26,10 +26,6 @@ import org.springframework.util.Assert;
*/
public class DispatcherPolicy {
public final static int DEFAULT_MAX_MESSAGES_PER_TASK = 1;
public final static long DEFAULT_RECEIVE_TIMEOUT = 1000;
public final static int DEFAULT_REJECTION_LIMIT = 5;
public final static long DEFAULT_RETRY_INTERVAL = 1000;
@@ -37,10 +33,6 @@ public class DispatcherPolicy {
private final boolean publishSubscribe;
private volatile int maxMessagesPerTask = DEFAULT_MAX_MESSAGES_PER_TASK;
private volatile long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
private volatile int rejectionLimit = DEFAULT_REJECTION_LIMIT;
private volatile long retryInterval = DEFAULT_RETRY_INTERVAL;
@@ -72,35 +64,6 @@ public class DispatcherPolicy {
return this.publishSubscribe;
}
/**
* Return the maximum number of messages for each retrieval attempt.
*/
public int getMaxMessagesPerTask() {
return this.maxMessagesPerTask;
}
/**
* Set the maximum number of messages for each retrieval attempt.
*/
public void setMaxMessagesPerTask(int maxMessagesPerTask) {
Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagePerTask' must be at least 1");
this.maxMessagesPerTask = maxMessagesPerTask;
}
/**
* Return the maximum amount of time in milliseconds to wait for a message to be available.
*/
public long getReceiveTimeout() {
return this.receiveTimeout;
}
/**
* Set the maximum amount of time in milliseconds to wait for a message to be available.
*/
public void setReceiveTimeout(long receiveTimeout) {
this.receiveTimeout = receiveTimeout;
}
/**
* Return the maximum number of retries upon rejection.
*/

2
org.springframework.integration/src/main/java/org/springframework/integration/channel/ThreadLocalChannel.java
View File

@@ -88,8 +88,6 @@ public class ThreadLocalChannel extends AbstractMessageChannel {
private static DispatcherPolicy defaultDispatcherPolicy() {
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(false);
dispatcherPolicy.setMaxMessagesPerTask(1);
dispatcherPolicy.setReceiveTimeout(0);
dispatcherPolicy.setRejectionLimit(1);
dispatcherPolicy.setRetryInterval(0);
dispatcherPolicy.setShouldFailOnRejectionLimit(false);

9
org.springframework.integration/src/main/java/org/springframework/integration/channel/config/AbstractChannelParser.java
View File

@@ -49,6 +49,7 @@ public abstract class AbstractChannelParser extends AbstractSingleBeanDefinition
private static final String INTERCEPTORS_PROPERTY = "interceptors";
@Override
protected boolean shouldGenerateId() {
return false;
@@ -104,14 +105,6 @@ public abstract class AbstractChannelParser extends AbstractSingleBeanDefinition
}
private void configureDispatcherPolicy(Element element, DispatcherPolicy dispatcherPolicy) {
String maxMessagesPerTask = element.getAttribute("max-messages-per-task");
if (StringUtils.hasText(maxMessagesPerTask)) {
dispatcherPolicy.setMaxMessagesPerTask(Integer.parseInt(maxMessagesPerTask));
}
String receiveTimeout = element.getAttribute("receive-timeout");
if (StringUtils.hasText(receiveTimeout)) {
dispatcherPolicy.setReceiveTimeout(Long.parseLong(receiveTimeout));
}
String rejectionLimit = element.getAttribute("rejection-limit");
if (StringUtils.hasText(rejectionLimit)) {
dispatcherPolicy.setRejectionLimit(Integer.parseInt(rejectionLimit));

2
org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd
View File

@@ -295,8 +295,6 @@
Defines a dispatcher policy.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="max-messages-per-task" type="xsd:int"/>
<xsd:attribute name="receive-timeout" type="xsd:long"/>
<xsd:attribute name="rejection-limit" type="xsd:int"/>
<xsd:attribute name="retry-interval" type="xsd:long"/>
<xsd:attribute name="should-fail-on-rejection-limit" type="xsd:boolean"/>

4
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/DirectChannel.java
View File

@@ -86,7 +86,7 @@ public class DirectChannel extends AbstractMessageChannel implements Subscribabl
@Override
protected boolean doSend(Message<?> message, long timeout) {
if (message != null && this.handlerCount.get() > 0) {
return this.dispatcher.dispatch(message);
return this.dispatcher.send(message);
}
return false;
}
@@ -102,8 +102,6 @@ public class DirectChannel extends AbstractMessageChannel implements Subscribabl
private static DispatcherPolicy defaultDispatcherPolicy() {
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(false);
dispatcherPolicy.setMaxMessagesPerTask(1);
dispatcherPolicy.setReceiveTimeout(0);
dispatcherPolicy.setRejectionLimit(1);
dispatcherPolicy.setRetryInterval(0);
dispatcherPolicy.setShouldFailOnRejectionLimit(false);

5
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java
View File

@@ -17,15 +17,16 @@
package org.springframework.integration.dispatcher;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageTarget;
/**
* Strategy interface for dispatching messages.
*
* @author Mark Fisher
*/
public interface MessageDispatcher {
public interface MessageDispatcher extends MessageTarget {
boolean dispatch(Message<?> message);
boolean send(Message<?> message);
void setSendTimeout(long timeout);

29
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcherTask.java
View File

@@ -37,6 +37,10 @@ public class PollingDispatcherTask implements SchedulableTask, Subscribable {
private final SimpleDispatcher dispatcher;
private volatile long receiveTimeout = -1;
private volatile int maxMessagesPerTask = 1;
public PollingDispatcherTask(MessageChannel channel, Schedule schedule) {
Assert.notNull(channel, "channel must not be null");
@@ -46,6 +50,22 @@ public class PollingDispatcherTask implements SchedulableTask, Subscribable {
}
/**
* Set the maximum amount of time in milliseconds to wait for a message to be available.
* A negative value indicates that receive calls should block indefinitely.
*/
public void setReceiveTimeout(long receiveTimeout) {
this.receiveTimeout = receiveTimeout;
}
/**
* Set the maximum number of messages for each retrieval attempt.
*/
public void setMaxMessagesPerTask(int maxMessagesPerTask) {
Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagePerTask' must be at least 1");
this.maxMessagesPerTask = maxMessagesPerTask;
}
public boolean subscribe(MessageTarget target) {
return this.dispatcher.subscribe(target);
}
@@ -59,15 +79,14 @@ public class PollingDispatcherTask implements SchedulableTask, Subscribable {
}
public void run() {
long timeout = this.channel.getDispatcherPolicy().getReceiveTimeout();
int limit = this.channel.getDispatcherPolicy().getMaxMessagesPerTask();
int count = 0;
while (count < limit) {
Message<?> message = (timeout < 0) ? this.channel.receive() : this.channel.receive(timeout);
while (count < this.maxMessagesPerTask) {
Message<?> message = (this.receiveTimeout < 0) ?
this.channel.receive() : this.channel.receive(this.receiveTimeout);
if (message == null) {
return;
}
this.dispatcher.dispatch(message);
this.dispatcher.send(message);
count++;
}
}

5
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java
View File

@@ -30,7 +30,6 @@ import org.springframework.integration.handler.MessageHandlerRejectedExecutionEx
import org.springframework.integration.message.BlockingTarget;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.Subscribable;
import org.springframework.integration.message.MessageTarget;
/**
@@ -38,7 +37,7 @@ import org.springframework.integration.message.MessageTarget;
*
* @author Mark Fisher
*/
public class SimpleDispatcher implements MessageDispatcher, Subscribable {
public class SimpleDispatcher implements MessageDispatcher {
protected final Log logger = LogFactory.getLog(this.getClass());
@@ -66,7 +65,7 @@ public class SimpleDispatcher implements MessageDispatcher, Subscribable {
return this.targets.remove(target);
}
public boolean dispatch(Message<?> message) {
public boolean send(Message<?> message) {
int attempts = 0;
List<MessageTarget> targetList = new ArrayList<MessageTarget>(this.targets);
while (attempts < this.dispatcherPolicy.getRejectionLimit()) {

2
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourceEndpoint.java
View File

@@ -67,7 +67,7 @@ public class SourceEndpoint extends AbstractEndpoint {
if (message == null) {
return false;
}
boolean sent = this.dispatcher.dispatch(message);
boolean sent = this.dispatcher.send(message);
if (this.source instanceof MessageDeliveryAware) {
if (sent) {
((MessageDeliveryAware) this.source).onSend(message);

8
org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java
View File

@@ -131,8 +131,6 @@ public class ChannelParserTests {
MessageChannel channel = (MessageChannel) context.getBean("pointToPointChannelByDefault");
DispatcherPolicy dispatcherPolicy = channel.getDispatcherPolicy();
assertFalse(dispatcherPolicy.isPublishSubscribe());
assertEquals(DispatcherPolicy.DEFAULT_MAX_MESSAGES_PER_TASK, dispatcherPolicy.getMaxMessagesPerTask());
assertEquals(DispatcherPolicy.DEFAULT_RECEIVE_TIMEOUT, dispatcherPolicy.getReceiveTimeout());
assertEquals(DispatcherPolicy.DEFAULT_REJECTION_LIMIT, dispatcherPolicy.getRejectionLimit());
assertEquals(DispatcherPolicy.DEFAULT_RETRY_INTERVAL, dispatcherPolicy.getRetryInterval());
assertTrue(dispatcherPolicy.getShouldFailOnRejectionLimit());
@@ -145,10 +143,8 @@ public class ChannelParserTests {
MessageChannel channel = (MessageChannel) context.getBean("channelWithDispatcherPolicy");
DispatcherPolicy dispatcherPolicy = channel.getDispatcherPolicy();
assertTrue(dispatcherPolicy.isPublishSubscribe());
assertEquals(7, dispatcherPolicy.getMaxMessagesPerTask());
assertEquals(77, dispatcherPolicy.getReceiveTimeout());
assertEquals(777, dispatcherPolicy.getRejectionLimit());
assertEquals(7777, dispatcherPolicy.getRetryInterval());
assertEquals(7, dispatcherPolicy.getRejectionLimit());
assertEquals(77, dispatcherPolicy.getRetryInterval());
assertFalse(dispatcherPolicy.getShouldFailOnRejectionLimit());
}

6
org.springframework.integration/src/test/java/org/springframework/integration/channel/config/channelParserTests.xml
View File

@@ -16,10 +16,8 @@
<channel id="publishSubscribeChannel" publish-subscribe="true"/>
<channel id="channelWithDispatcherPolicy" publish-subscribe="true">
<dispatcher-policy max-messages-per-task="7"
receive-timeout="77"
rejection-limit="777"
retry-interval="7777"
<dispatcher-policy rejection-limit="7"
retry-interval="77"
should-fail-on-rejection-limit="false"/>
</channel>

5
org.springframework.integration/src/test/java/org/springframework/integration/channel/factory/ChannelFactoryTests.java
View File

@@ -60,11 +60,6 @@ public class ChannelFactoryTests {
interceptors.add(new TestChannelInterceptor());
}
@Before
public void initDispatcherPolicy() {
dispatcherPolicy.setMaxMessagesPerTask(100);
}
@Test
public void testQueueChannelFactory() {

6
org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java
View File

@@ -41,7 +41,7 @@ public class SimpleDispatcherTests {
SimpleDispatcher dispatcher = new SimpleDispatcher(new DispatcherPolicy());
final CountDownLatch latch = new CountDownLatch(1);
dispatcher.subscribe(createEndpoint(TestHandlers.countDownHandler(latch)));
dispatcher.dispatch(new StringMessage("test"));
dispatcher.send(new StringMessage("test"));
latch.await(500, TimeUnit.MILLISECONDS);
assertEquals(0, latch.getCount());
}
@@ -54,7 +54,7 @@ public class SimpleDispatcherTests {
final AtomicInteger counter2 = new AtomicInteger();
dispatcher.subscribe(createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch)));
dispatcher.subscribe(createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch)));
dispatcher.dispatch(new StringMessage("test"));
dispatcher.send(new StringMessage("test"));
latch.await(500, TimeUnit.MILLISECONDS);
assertEquals(0, latch.getCount());
assertEquals("only 1 handler should have received the message", 1, counter1.get() + counter2.get());
@@ -68,7 +68,7 @@ public class SimpleDispatcherTests {
final AtomicInteger counter2 = new AtomicInteger();
dispatcher.subscribe(createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch)));
dispatcher.subscribe(createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch)));
dispatcher.dispatch(new StringMessage("test"));
dispatcher.send(new StringMessage("test"));
latch.await(500, TimeUnit.MILLISECONDS);
assertEquals(0, latch.getCount());
assertEquals(1, counter1.get());