Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

applied patch for http://issues.apache.org/activemq/browse/AMQ-1006 

git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@477066 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonas B. Lim 2006-11-20 06:20:12 +00:00
parent 979c5d55ee
commit b02ca01826
2 changed files with 32 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
activemq-core/src/main/java/org/apache/activemq/broker/region/policy/RoundRobinDispatchPolicy.java
View File

@ -53,6 +53,7 @@ public class RoundRobinDispatchPolicy implements DispatchPolicy {
synchronized(consumers) {
int count = 0;
Subscription firstMatchingConsumer = null;
for (Iterator iter = consumers.iterator(); iter.hasNext();) {
Subscription sub = (Subscription) iter.next();
@ -60,16 +61,20 @@ public class RoundRobinDispatchPolicy implements DispatchPolicy {
if (!sub.matches(node, msgContext))
continue;
if (firstMatchingConsumer == null) {
firstMatchingConsumer = sub;
}
sub.add(node);
count++;
}
if (firstMatchingConsumer != null) {
// Rotate the consumer list.
try {
if (consumers.size() > 1)
consumers.add(consumers.remove(0));
} catch (Throwable bestEffort) {
log.error("Caught error rotating consumers",bestEffort);
consumers.remove(firstMatchingConsumer);
consumers.add(firstMatchingConsumer);
} catch (Throwable bestEffort) { }
}
return count > 0;
}

22
activemq-core/src/test/java/org/apache/activemq/broker/policy/RoundRobinDispatchPolicyTest.java
View File

@ -23,6 +23,11 @@ import org.apache.activemq.broker.region.policy.PolicyEntry;
import org.apache.activemq.broker.region.policy.RoundRobinDispatchPolicy;
import org.apache.activemq.broker.region.policy.PolicyMap;
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.MessageConsumer;
import javax.jms.Session;
public class RoundRobinDispatchPolicyTest extends QueueSubscriptionTest {
protected BrokerService createBroker() throws Exception {
@ -82,6 +87,23 @@ public class RoundRobinDispatchPolicyTest extends QueueSubscriptionTest {
assertMessagesDividedAmongConsumers();
}
public void testOneProducerTwoMatchingConsumersOneNotMatchingConsumer() throws Exception {
// Create consumer that won't consume any message
createMessageConsumer(createConnectionFactory().createConnection(), createDestination(), "JMSPriority<1");
super.testOneProducerTwoConsumersSmallMessagesLargePrefetch();
assertMessagesDividedAmongConsumers();
}
protected MessageConsumer createMessageConsumer(Connection conn, Destination dest, String selector) throws Exception {
connections.add(conn);
Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
final MessageConsumer consumer = sess.createConsumer(dest, selector);
conn.start();
return consumer;
}
public void assertMessagesDividedAmongConsumers() {
assertEachConsumerReceivedAtLeastXMessages((messageCount * producerCount) / consumerCount);
assertEachConsumerReceivedAtMostXMessages(((messageCount * producerCount) / consumerCount) + 1);