You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ra...@apache.org on 2010/07/04 08:28:36 UTC
svn commit: r960298 - in /activemq/trunk/activemq-core/src:
main/java/org/apache/activemq/broker/region/
main/java/org/apache/activemq/broker/region/policy/
test/java/org/apache/activemq/ test/resources/org/apache/activemq/broker/
Author: rajdavies
Date: Sun Jul 4 06:28:36 2010
New Revision: 960298
URL: http://svn.apache.org/viewvc?rev=960298&view=rev
Log:
Applied patch for https://issues.apache.org/activemq/browse/AMQ-2795
Added:
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/ExclusiveConsumerStartupDestinationTest.java (with props)
activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/exclusive-consumer-startup-destination.xml (with props)
Modified:
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java?rev=960298&r1=960297&r2=960298&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java Sun Jul 4 06:28:36 2010
@@ -116,6 +116,7 @@ public class Queue extends BaseDestinati
private int consumersBeforeDispatchStarts = 0;
private CountDownLatch consumersBeforeStartsLatch;
private final AtomicLong pendingWakeups = new AtomicLong();
+ private boolean allConsumersExclusiveByDefault = false;
private final Runnable sendMessagesWaitingForSpaceTask = new Runnable() {
public void run() {
@@ -368,7 +369,7 @@ public class Queue extends BaseDestinati
}
addToConsumerList(sub);
- if (sub.getConsumerInfo().isExclusive()) {
+ if (sub.getConsumerInfo().isExclusive() || isAllConsumersExclusiveByDefault()) {
Subscription exclusiveConsumer = dispatchSelector.getExclusiveConsumer();
if (exclusiveConsumer == null) {
exclusiveConsumer = sub;
@@ -428,6 +429,16 @@ public class Queue extends BaseDestinati
}
dispatchSelector.setExclusiveConsumer(exclusiveConsumer);
}
+ } else if (isAllConsumersExclusiveByDefault()) {
+ Subscription exclusiveConsumer = null;
+ for (Subscription s : consumers) {
+ if (exclusiveConsumer == null
+ || s.getConsumerInfo().getPriority() > exclusiveConsumer
+ .getConsumerInfo().getPriority()) {
+ exclusiveConsumer = s;
+ }
+ }
+ dispatchSelector.setExclusiveConsumer(exclusiveConsumer);
}
ConsumerId consumerId = sub.getConsumerInfo().getConsumerId();
getMessageGroupOwners().removeConsumer(consumerId);
@@ -881,6 +892,15 @@ public class Queue extends BaseDestinati
this.consumersBeforeDispatchStarts = consumersBeforeDispatchStarts;
}
+ public void setAllConsumersExclusiveByDefault(boolean allConsumersExclusiveByDefault) {
+ this.allConsumersExclusiveByDefault = allConsumersExclusiveByDefault;
+ }
+
+ public boolean isAllConsumersExclusiveByDefault() {
+ return allConsumersExclusiveByDefault;
+ }
+
+
// Implementation methods
// -------------------------------------------------------------------------
private QueueMessageReference createMessageReference(Message message) {
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java?rev=960298&r1=960297&r2=960298&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java Sun Jul 4 06:28:36 2010
@@ -88,6 +88,7 @@ public class PolicyEntry extends Destina
private int storeUsageHighWaterMark = 100;
private SlowConsumerStrategy slowConsumerStrategy;
private boolean prioritizedMessages;
+ private boolean allConsumersExclusiveByDefault;
public void configure(Broker broker,Queue queue) {
@@ -111,6 +112,7 @@ public class PolicyEntry extends Destina
queue.setLazyDispatch(isLazyDispatch());
queue.setTimeBeforeDispatchStarts(getTimeBeforeDispatchStarts());
queue.setConsumersBeforeDispatchStarts(getConsumersBeforeDispatchStarts());
+ queue.setAllConsumersExclusiveByDefault(isAllConsumersExclusiveByDefault());
}
public void configure(Broker broker,Topic topic) {
@@ -751,4 +753,12 @@ public class PolicyEntry extends Destina
this.prioritizedMessages = prioritizedMessages;
}
+ public void setAllConsumersExclusiveByDefault(boolean allConsumersExclusiveByDefault) {
+ this.allConsumersExclusiveByDefault = allConsumersExclusiveByDefault;
+ }
+
+ public boolean isAllConsumersExclusiveByDefault() {
+ return allConsumersExclusiveByDefault;
+ }
+
}
Added: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/ExclusiveConsumerStartupDestinationTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/ExclusiveConsumerStartupDestinationTest.java?rev=960298&view=auto
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/ExclusiveConsumerStartupDestinationTest.java (added)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/ExclusiveConsumerStartupDestinationTest.java Sun Jul 4 06:28:36 2010
@@ -0,0 +1,206 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.activemq;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import junit.framework.Assert;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.apache.activemq.command.ActiveMQQueue;
+
+public class ExclusiveConsumerStartupDestinationTest extends EmbeddedBrokerTestSupport{
+
+ private static final String VM_BROKER_URL = "vm://localhost";
+
+ protected BrokerService createBroker() throws Exception {
+ BrokerService answer = new BrokerService();
+ answer.setPersistent(false);
+ PolicyMap map = new PolicyMap();
+ PolicyEntry entry = new PolicyEntry();
+ entry.setAllConsumersExclusiveByDefault(true);
+ map.setDefaultEntry(entry);
+ answer.setDestinationPolicy(map);
+ return answer;
+ }
+
+ protected String getBrokerConfigUri() {
+ return "org/apache/activemq/broker/exclusive-consumer-startup-destination.xml";
+ }
+
+ private Connection createConnection(final boolean start) throws JMSException {
+ ConnectionFactory cf = new ActiveMQConnectionFactory(VM_BROKER_URL);
+ Connection conn = cf.createConnection();
+ if (start) {
+ conn.start();
+ }
+ return conn;
+ }
+
+ public void testExclusiveConsumerSelectedCreatedFirst() throws JMSException, InterruptedException {
+ Connection conn = createConnection(true);
+
+ Session exclusiveSession = null;
+ Session fallbackSession = null;
+ Session senderSession = null;
+
+ try {
+
+ exclusiveSession = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ fallbackSession = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ senderSession = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+ ActiveMQQueue exclusiveQueue = new ActiveMQQueue("TEST.QUEUE1");
+ MessageConsumer exclusiveConsumer = exclusiveSession.createConsumer(exclusiveQueue);
+
+ ActiveMQQueue fallbackQueue = new ActiveMQQueue("TEST.QUEUE1");
+ MessageConsumer fallbackConsumer = fallbackSession.createConsumer(fallbackQueue);
+
+ ActiveMQQueue senderQueue = new ActiveMQQueue("TEST.QUEUE1");
+
+ MessageProducer producer = senderSession.createProducer(senderQueue);
+
+ Message msg = senderSession.createTextMessage("test");
+ producer.send(msg);
+ // TODO need two send a 2nd message - bug AMQ-1024
+ // producer.send(msg);
+ Thread.sleep(100);
+
+ // Verify exclusive consumer receives the message.
+ Assert.assertNotNull(exclusiveConsumer.receive(100));
+ Assert.assertNull(fallbackConsumer.receive(100));
+
+ } finally {
+ fallbackSession.close();
+ senderSession.close();
+ conn.close();
+ }
+
+ }
+
+ public void testFailoverToAnotherExclusiveConsumerCreatedFirst() throws JMSException,
+ InterruptedException {
+ Connection conn = createConnection(true);
+
+ Session exclusiveSession1 = null;
+ Session exclusiveSession2 = null;
+ Session fallbackSession = null;
+ Session senderSession = null;
+
+ try {
+
+ exclusiveSession1 = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ exclusiveSession2 = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ fallbackSession = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ senderSession = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+ // This creates the exclusive consumer first which avoids AMQ-1024
+ // bug.
+ ActiveMQQueue exclusiveQueue = new ActiveMQQueue("TEST.QUEUE2");
+ MessageConsumer exclusiveConsumer1 = exclusiveSession1.createConsumer(exclusiveQueue);
+ MessageConsumer exclusiveConsumer2 = exclusiveSession2.createConsumer(exclusiveQueue);
+
+ ActiveMQQueue fallbackQueue = new ActiveMQQueue("TEST.QUEUE2");
+ MessageConsumer fallbackConsumer = fallbackSession.createConsumer(fallbackQueue);
+
+ ActiveMQQueue senderQueue = new ActiveMQQueue("TEST.QUEUE2");
+
+ MessageProducer producer = senderSession.createProducer(senderQueue);
+
+ Message msg = senderSession.createTextMessage("test");
+ producer.send(msg);
+ Thread.sleep(100);
+
+ // Verify exclusive consumer receives the message.
+ Assert.assertNotNull(exclusiveConsumer1.receive(100));
+ Assert.assertNull(exclusiveConsumer2.receive(100));
+ Assert.assertNull(fallbackConsumer.receive(100));
+
+ // Close the exclusive consumer to verify the non-exclusive consumer
+ // takes over
+ exclusiveConsumer1.close();
+
+ producer.send(msg);
+ producer.send(msg);
+
+ Assert.assertNotNull("Should have received a message", exclusiveConsumer2.receive(100));
+ Assert.assertNull("Should not have received a message", fallbackConsumer.receive(100));
+
+ } finally {
+ fallbackSession.close();
+ senderSession.close();
+ conn.close();
+ }
+
+ }
+
+ public void testFailoverToNonExclusiveConsumer() throws JMSException, InterruptedException {
+ Connection conn = createConnection(true);
+
+ Session exclusiveSession = null;
+ Session fallbackSession = null;
+ Session senderSession = null;
+
+ try {
+
+ exclusiveSession = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ fallbackSession = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ senderSession = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+ // This creates the exclusive consumer first which avoids AMQ-1024
+ // bug.
+ ActiveMQQueue exclusiveQueue = new ActiveMQQueue("TEST.QUEUE3");
+ MessageConsumer exclusiveConsumer = exclusiveSession.createConsumer(exclusiveQueue);
+
+ ActiveMQQueue fallbackQueue = new ActiveMQQueue("TEST.QUEUE3");
+ MessageConsumer fallbackConsumer = fallbackSession.createConsumer(fallbackQueue);
+
+ ActiveMQQueue senderQueue = new ActiveMQQueue("TEST.QUEUE3");
+
+ MessageProducer producer = senderSession.createProducer(senderQueue);
+
+ Message msg = senderSession.createTextMessage("test");
+ producer.send(msg);
+ Thread.sleep(100);
+
+ // Verify exclusive consumer receives the message.
+ Assert.assertNotNull(exclusiveConsumer.receive(100));
+ Assert.assertNull(fallbackConsumer.receive(100));
+
+ // Close the exclusive consumer to verify the non-exclusive consumer
+ // takes over
+ exclusiveConsumer.close();
+
+ producer.send(msg);
+
+ Assert.assertNotNull(fallbackConsumer.receive(100));
+
+ } finally {
+ fallbackSession.close();
+ senderSession.close();
+ conn.close();
+ }
+
+ }
+
+}
Propchange: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/ExclusiveConsumerStartupDestinationTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/ExclusiveConsumerStartupDestinationTest.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/exclusive-consumer-startup-destination.xml
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/exclusive-consumer-startup-destination.xml?rev=960298&view=auto
==============================================================================
--- activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/exclusive-consumer-startup-destination.xml (added)
+++ activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/exclusive-consumer-startup-destination.xml Sun Jul 4 06:28:36 2010
@@ -0,0 +1,46 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You 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.
+-->
+
+<!-- this file can only be parsed using the xbean-spring library -->
+<!-- START SNIPPET: xbean -->
+<beans
+ xmlns="http://www.springframework.org/schema/beans"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
+ http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd
+ http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">
+
+ <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer" />
+
+ <broker xmlns="http://activemq.apache.org/schema/core">
+ <destinationPolicy>
+ <policyMap>
+ <policyEntries>
+ <policyEntry queue="TEST.>" allConsumersExclusiveByDefault="true"/>
+ </policyEntries>
+ </policyMap>
+ </destinationPolicy>
+ <destinations>
+ <queue physicalName="TEST.QUEUE1"/>
+ <queue physicalName="TEST.QUEUE2"/>
+ <queue physicalName="TEST.QUEUE3"/>
+ </destinations>
+ </broker>
+
+</beans>
+<!-- END SNIPPET: xbean -->
Propchange: activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/exclusive-consumer-startup-destination.xml
------------------------------------------------------------------------------
svn:mime-type = text/plain