diff options
Diffstat (limited to 'java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessage.java')
| -rw-r--r-- | java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessage.java | 23 |
1 files changed, 18 insertions, 5 deletions
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 df7cecc940..0112d3b388 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 class UnacknowledgedMessage 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 class UnacknowledgedMessage 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() |