commit d79a22bba1c8bffcdd0129a7e72d4d692351915c
author Aidan Skinner <aidan@apache.org> Mon May 05 10:24:50 2008 +0000
committer Aidan Skinner <aidan@apache.org> Mon May 05 10:24:50 2008 +0000
tree a8dfb19331709ac4b0ab75ecc22a3954ccc05a95
parent d9ceec402450d18ffd363e45f32f9e6a301b57c9

QPID-1019 prevent messages being dequeued unecessarily, from rgodfrey

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/M2.x@653416 13f79535-47bb-0310-9956-ffa450edef68

java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java

diff --git a/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java b/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
index 74169a1..1314b2b 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
@@ -469,7 +469,7 @@
 
         synchronized (_unacknowledgedMessageMap.getLock())
         {
-            _unacknowledgedMessageMap.add(deliveryTag, new UnacknowledgedMessage(entry, consumerTag, deliveryTag));
+            _unacknowledgedMessageMap.add(deliveryTag, new UnacknowledgedMessage(entry, consumerTag, deliveryTag,_unacknowledgedMessageMap));
             checkSuspension();
         }
     }

java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessage.java

diff --git a/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessage.java b/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessage.java
index df7cecc..0112d3b 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessage.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessage.java
@@ -34,13 +34,18 @@
     public final long deliveryTag;
 
     private boolean _queueDeleted;
+    private final UnacknowledgedMessageMap _unacknowledgeMessageMap;
 
 
-    public UnacknowledgedMessage(QueueEntry entry, AMQShortString consumerTag, long deliveryTag)
+    public UnacknowledgedMessage(QueueEntry entry,
+                                 AMQShortString consumerTag,
+                                 long deliveryTag,
+                                 final UnacknowledgedMessageMap unacknowledgedMessageMap)
     {
         this.entry = entry;
         this.consumerTag = consumerTag;
         this.deliveryTag = deliveryTag;
+        _unacknowledgeMessageMap = unacknowledgedMessageMap;
     }
 
     public String toString()
@@ -60,12 +65,20 @@
 
     public void discard(StoreContext storeContext) throws AMQException
     {
-        if (entry.getQueue() != null)
+        synchronized(_unacknowledgeMessageMap)
         {
-            entry.getQueue().dequeue(storeContext, entry);
+            if(_unacknowledgeMessageMap.contains(deliveryTag))
+            {
+
+                if (entry.getQueue() != null)
+                {
+                    entry.getQueue().dequeue(storeContext, entry);
+                }
+                //if the queue is null then the message is waiting to be acked, but has been removed.
+                entry.getMessage().decrementReference(storeContext);
+            }
         }
-        //if the queue is null then the message is waiting to be acked, but has been removed.
-        entry.getMessage().decrementReference(storeContext);
+        
     }
 
     public AMQMessage getMessage()

java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java

diff --git a/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java b/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java
index 7b8db75..1b4187b 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java
@@ -137,7 +137,7 @@
                 };
 
                 TestMessage message = new TestMessage(deliveryTag, i, info, txnContext);
-                _map.add(deliveryTag, new UnacknowledgedMessage(new QueueEntry(null,message), null, deliveryTag));
+                _map.add(deliveryTag, new UnacknowledgedMessage(new QueueEntry(null,message), null, deliveryTag, _map));
             }
             _acked = acked;
             _unacked = unacked;