QPID-3927: ensure that priority is properly accounted for when comparing messages on different QueueEntryLists contained within the encompassing PriorityQueue

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

2 files changed, 48 insertions, 4 deletions

diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/PriorityQueueList.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/PriorityQueueList.java
index 05141a48a1..1d13ee66c0 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/PriorityQueueList.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/PriorityQueueList.java
@@ -25,19 +25,19 @@ import org.apache.qpid.server.message.ServerMessage;
 public class PriorityQueueList implements QueueEntryList<SimpleQueueEntryImpl>
 {
     private final AMQQueue _queue;
-    private final SimpleQueueEntryList[] _priorityLists;
+    private final PriorityQueueEntrySubList[] _priorityLists;
     private final int _priorities;
     private final int _priorityOffset;
 
     public PriorityQueueList(AMQQueue queue, int priorities)
     {
         _queue = queue;
-        _priorityLists = new SimpleQueueEntryList[priorities];
+        _priorityLists = new PriorityQueueEntrySubList[priorities];
         _priorities = priorities;
         _priorityOffset = 5-((priorities + 1)/2);
         for(int i = 0; i < priorities; i++)
         {
-            _priorityLists[i] = new SimpleQueueEntryList(queue);
+            _priorityLists[i] = new PriorityQueueEntrySubList(queue);
         }
     }
 
@@ -161,4 +161,48 @@ public class PriorityQueueList implements QueueEntryList<SimpleQueueEntryImpl>
             return new PriorityQueueList(queue, _priorities);
         }
     }
+
+    private static class PriorityQueueEntrySubList extends SimpleQueueEntryList
+    {
+        public PriorityQueueEntrySubList(AMQQueue queue)
+        {
+            super(queue);
+        }
+
+        @Override
+        protected PriorityQueueEntryImpl createQueueEntry(ServerMessage<?> message)
+        {
+            return new PriorityQueueEntryImpl(this, message);
+        }
+    }
+
+    private static class PriorityQueueEntryImpl extends SimpleQueueEntryImpl
+    {
+        public PriorityQueueEntryImpl(PriorityQueueEntrySubList queueEntryList, ServerMessage<?> message)
+        {
+            super(queueEntryList, message);
+        }
+
+        @Override
+        public int compareTo(final QueueEntry o)
+        {
+            byte thisPriority = getMessageHeader().getPriority();
+            byte otherPriority = o.getMessageHeader().getPriority();
+
+            if(thisPriority != otherPriority)
+            {
+                /*
+                 * Different priorities, so answer can only be greater than or less than
+                 *
+                 * A message with higher priority (e.g. 5) is conceptually 'earlier' in the
+                 * priority queue than one with a lower priority (e.g. 4).
+                 */
+                return thisPriority > otherPriority ? -1 : 1;
+            }
+            else
+            {
+                return super.compareTo(o);
+            }
+        }
+    }
 }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleQueueEntryList.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleQueueEntryList.java
index c82d1b984a..b8d8ec19f4 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleQueueEntryList.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleQueueEntryList.java
@@ -104,7 +104,7 @@ public class SimpleQueueEntryList implements QueueEntryList<SimpleQueueEntryImpl
         }
     }
 
-    protected SimpleQueueEntryImpl createQueueEntry(ServerMessage message)
+    protected SimpleQueueEntryImpl createQueueEntry(ServerMessage<?> message)
     {
         return new SimpleQueueEntryImpl(this, message);
     }