activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rajdav...@apache.org
Subject 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/
Date Sun, 04 Jul 2010 06:28:36 GMT
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



Mime
View raw message