QPID-6190 : [Java Broker] [AMQP 1.0] Fix transactional sending to QueueDestination

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1634645 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 12 insertions, 5 deletions

diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/QueueDestination.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/QueueDestination.java
index 4fefbe9f7d..4d26890e72 100644
--- a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/QueueDestination.java
+++ b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/QueueDestination.java
@@ -21,12 +21,11 @@
 package org.apache.qpid.server.protocol.v1_0;
 
 import org.apache.log4j.Logger;
+
 import org.apache.qpid.amqp_1_0.type.Outcome;
 import org.apache.qpid.amqp_1_0.type.messaging.Accepted;
-
-import org.apache.qpid.server.message.MessageSource;
+import org.apache.qpid.server.message.MessageReference;
 import org.apache.qpid.server.queue.AMQQueue;
-
 import org.apache.qpid.server.txn.ServerTransaction;
 
 public class QueueDestination extends MessageSourceDestination implements SendingDestination, ReceivingDestination
@@ -51,16 +50,24 @@ public class QueueDestination extends MessageSourceDestination implements Sendin
 
         txn.enqueue(getQueue(),message, new ServerTransaction.Action()
         {
+            MessageReference _reference = message.newReference();
 
 
             public void postCommit()
             {
-                getQueue().enqueue(message,null);
+                try
+                {
+                    getQueue().enqueue(message, null);
+                }
+                finally
+                {
+                    _reference.release();
+                }
             }
 
             public void onRollback()
             {
-                // NO-OP
+                _reference.release();
             }
         });