From 01d15f394fba150f5d641819df8601539e671535 Mon Sep 17 00:00:00 2001 From: Martin Ritchie Date: Fri, 6 Mar 2009 12:29:27 +0000 Subject: QPID-1635 : Feed back from review, ensure that getMessage() never returns null. Used setDeliveredToConsumers() on QEI to know when it is safe to unload message so SAMQQueue does not need to have anything to do with message un/loading. So the FtD work is fully encapuslated by the QEI which delegates memory accounting to its list. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@750874 13f79535-47bb-0310-9956-ffa450edef68 --- .../java/org/apache/qpid/server/queue/QueueEntryImpl.java | 10 ++++++++++ .../java/org/apache/qpid/server/queue/SimpleAMQQueue.java | 15 --------------- 2 files changed, 10 insertions(+), 15 deletions(-) (limited to 'qpid/java') diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java index c510ec3374..5114529419 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java @@ -133,6 +133,10 @@ public class QueueEntryImpl implements QueueEntry public AMQMessage getMessage() { + if (_message == null) + { + return _backingStore.load(_messageId); + } return _message; } @@ -149,6 +153,12 @@ public class QueueEntryImpl implements QueueEntry public void setDeliveredToConsumer() { _flags |= DELIVERED_TO_CONSUMER; + + // We have delivered this message so we can unload it if we are flowed. + if (_queueEntryList.isFlowed()) + { + unload(); + } } public boolean expired() throws AMQException diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java index 5730e419d5..9e9895c53b 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java @@ -466,23 +466,8 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener { _deliveredMessages.incrementAndGet(); - if (entry.isFlowed()) - { - if(_logger.isDebugEnabled()) - { - _logger.debug("Synchoronus load of entry:" + entry.debugIdentity()); - } - entry.load(); - } - sub.send(entry); - // We have delivered this message so we can unload it. - if (_entries.isFlowed() && entry.isAcquired() && entry.getDeliveredToConsumer()) - { - entry.unload(); - } - } private boolean subscriptionReadyAndHasInterest(final Subscription sub, final QueueEntry entry) -- cgit v1.2.1