merge from trunk

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/QPID-6262-JavaBrokerNIO@1658748 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 38 insertions, 14 deletions

diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java
index b5e1bdafbb..a44768ffdc 100644
--- a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java
+++ b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java
@@ -347,6 +347,11 @@ class ConsumerTarget_1_0 extends AbstractConsumerTarget
         return _link.getSession();
     }
 
+    public void flush()
+    {
+        _consumer.flush();
+    }
+
     private class DispositionAction implements UnsettledAction
     {
 
diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Message_1_0.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Message_1_0.java
index 36796851e0..18f5ba9e2e 100644
--- a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Message_1_0.java
+++ b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Message_1_0.java
@@ -21,23 +21,42 @@
 package org.apache.qpid.server.protocol.v1_0;
 
 
+import java.lang.ref.SoftReference;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
+
 import org.apache.qpid.server.message.AbstractServerMessageImpl;
 import org.apache.qpid.server.store.StoredMessage;
 
 public class Message_1_0 extends AbstractServerMessageImpl<Message_1_0, MessageMetaData_1_0>
 {
 
-    private List<ByteBuffer> _fragments;
+    private volatile SoftReference<List<ByteBuffer>> _fragmentsRef;
     private long _arrivalTime;
+    private final long _size;
 
 
     public Message_1_0(final StoredMessage<MessageMetaData_1_0> storedMessage)
     {
         super(storedMessage, null);
-        _fragments = restoreFragments(storedMessage);
+        final List<ByteBuffer> fragments = restoreFragments(getStoredMessage());
+        _fragmentsRef = new SoftReference<>(fragments);
+        _size = calculateSize(fragments);
+    }
+
+    private long calculateSize(final List<ByteBuffer> fragments)
+    {
+
+        long size = 0l;
+        if(fragments != null)
+        {
+            for(ByteBuffer buf : fragments)
+            {
+                size += buf.remaining();
+            }
+        }
+        return size;
     }
 
     private static List<ByteBuffer> restoreFragments(StoredMessage<MessageMetaData_1_0> storedMessage)
@@ -65,7 +84,8 @@ public class Message_1_0 extends AbstractServerMessageImpl<Message_1_0, MessageM
                        final Object connectionReference)
     {
         super(storedMessage, connectionReference);
-        _fragments = fragments;
+        _fragmentsRef = new SoftReference<>(fragments);
+        _size = calculateSize(fragments);
         _arrivalTime = System.currentTimeMillis();
     }
 
@@ -94,16 +114,7 @@ public class Message_1_0 extends AbstractServerMessageImpl<Message_1_0, MessageM
 
     public long getSize()
     {
-        long size = 0l;
-        if(_fragments != null)
-        {
-            for(ByteBuffer buf : _fragments)
-            {
-                size += buf.remaining();
-            }
-        }
-
-        return size;
+        return _size;
     }
 
     public long getExpiration()
@@ -118,7 +129,14 @@ public class Message_1_0 extends AbstractServerMessageImpl<Message_1_0, MessageM
 
     public List<ByteBuffer> getFragments()
     {
-        return _fragments;
+
+        List<ByteBuffer> fragments = _fragmentsRef.get();
+        if(fragments == null)
+        {
+            fragments = restoreFragments(getStoredMessage());
+            _fragmentsRef = new SoftReference<>(fragments);
+        }
+        return fragments;
     }
 
 }
diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java
index f8e4853099..48ff420965 100644
--- a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java
+++ b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java
@@ -496,6 +496,7 @@ public class SendingLink_1_0 implements SendingLinkListener, Link_1_0, DeliveryS
                 && hasCredit())
         {
             _draining = true;
+            _target.flush();
         }
 
         while(!_resumeAcceptedTransfers.isEmpty() && getEndpoint().hasCreditToSend())