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/qpid@750874 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 10 insertions, 15 deletions

diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java b/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java
index c510ec3374..5114529419 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java
+++ b/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/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java b/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
index 5730e419d5..9e9895c53b 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
+++ b/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)