https://issues.apache.org/activemq/browse/AMQ-1024

- Fix case where a non-exclusive consumer registered first will receive a message, before an exclusive consumer.
- Added junit test case to demonstrate the issue.

git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@473219 13f79535-47bb-0310-9956-ffa450edef68

activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java

@@ -160,7 +160,12 @@ public class Queue implements Destination {
         MessageEvaluationContext msgContext = context.getMessageEvaluationContext();
         try {
             synchronized (consumers) {
-                consumers.add(sub);
+                if (sub.getConsumerInfo().isExclusive()) {
+                    // Add to front of list to ensure that an exclusive consumer gets all messages
+                    // before non-exclusive consumers
+                    consumers.add(0, sub);
+                } else {
+                    consumers.add(sub);
             }
 
             highestSubscriptionPriority = calcHighestSubscriptionPriority();

activemq-core/src/test/java/org/apache/activemq/JMSExclusiveConsumerTest.java

@@ -128,4 +128,29 @@ public class JMSExclusiveConsumerTest extends JmsTestSupport {
         assertNull(consumer1.receiveNoWait());
         assertNull(consumer2.receiveNoWait());
     }
+
+    public void testMixExclusiveWithNonExclusive() throws Exception {
+        ActiveMQQueue exclusiveQueue = new ActiveMQQueue("TEST.FOO?consumer.exclusive=true");
+        ActiveMQQueue nonExclusiveQueue = new ActiveMQQueue("TEST.FOO?consumer.exclusive=false");
+
+        connection.start();
+        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+        MessageConsumer nonExCon = session.createConsumer(nonExclusiveQueue);
+        MessageConsumer exCon = session.createConsumer(exclusiveQueue);
+
+
+        MessageProducer prod = session.createProducer(exclusiveQueue);
+        prod.send(session.createMessage());
+        prod.send(session.createMessage());
+        prod.send(session.createMessage());
+
+        Message m;
+        for (int i=0; i<3; i++) {
+            m = exCon.receive(1000);
+            assertNotNull(m);
+            m = nonExCon.receive(1000);
+            assertNull(m);
+        }
+    }
 }