QPID-1144 : Reference count drops to zero too early for immediate messages in a txn

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

2 files changed, 12 insertions, 4 deletions

diff --git a/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java b/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java
index ad8303ec5d..3c71282c57 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java
@@ -93,11 +93,19 @@ public class LocalTransactionalContext implements TransactionalContext
         public void process() throws AMQException
         {
 
-            QueueEntry entry = _queue.enqueue(getStoreContext(),_message);
+            _message.incrementReference();
+            try
+            {
+                QueueEntry entry = _queue.enqueue(getStoreContext(),_message);
 
-            if(entry.immediateAndNotDelivered())
+                if(entry.immediateAndNotDelivered())
+                {
+                    getReturnMessages().add(new NoConsumersException(_message));
+                }
+            }
+            finally
             {
-                getReturnMessages().add(new NoConsumersException(_message));
+                _message.decrementReference(getStoreContext());
             }
         }
     }
diff --git a/java/common/src/main/java/org/apache/qpid/AMQChannelException.java b/java/common/src/main/java/org/apache/qpid/AMQChannelException.java
index f6f596da95..ef9420ba87 100644
--- a/java/common/src/main/java/org/apache/qpid/AMQChannelException.java
+++ b/java/common/src/main/java/org/apache/qpid/AMQChannelException.java
@@ -54,6 +54,6 @@ public class AMQChannelException extends AMQException
     public AMQFrame getCloseFrame(int channel)
     {
         MethodRegistry reg = MethodRegistry.getMethodRegistry(new ProtocolVersion(major,minor));
-        return new AMQFrame(channel, reg.createChannelCloseBody(getErrorCode().getCode(), new AMQShortString(getMessage()),_classId,_methodId));
+        return new AMQFrame(channel, reg.createChannelCloseBody(getErrorCode() == null ? AMQConstant.INTERNAL_ERROR.getCode() : getErrorCode().getCode(), new AMQShortString(getMessage()),_classId,_methodId));
     }
 }