Ensure message reference is always released, even if there is an exception in processing the ImmmediateAction

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/java-broker-amqp-1-0-management@1565222 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 35 insertions, 29 deletions

diff --git a/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java b/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
index 041e106a40..d46cc8c72b 100644
--- a/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
+++ b/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
@@ -1208,41 +1208,47 @@ public class AMQChannel implements AMQSessionModel, AsyncAutoCommitTransaction.F
                 ServerTransaction txn = new LocalTransaction(_messageStore);
                 final AMQMessage message = (AMQMessage) entry.getMessage();
                 MessageReference ref = message.newReference();
-                entry.delete();
-                txn.dequeue(queue, message,
-                            new ServerTransaction.Action()
-                            {
-                                @Override
-                                public void postCommit()
+                try
+                {
+                    entry.delete();
+                    txn.dequeue(queue, message,
+                                new ServerTransaction.Action()
                                 {
-                                    try
-                                    {
-                                        final
-                                        ProtocolOutputConverter outputConverter =
-                                                _session.getProtocolOutputConverter();
-
-                                        outputConverter.writeReturn(message.getMessagePublishInfo(),
-                                                                    message.getContentHeaderBody(),
-                                                                    message,
-                                                                    _channelId,
-                                                                    AMQConstant.NO_CONSUMERS.getCode(),
-                                                                    IMMEDIATE_DELIVERY_REPLY_TEXT);
-                                    }
-                                    catch (AMQException e)
+                                    @Override
+                                    public void postCommit()
                                     {
-                                        throw new RuntimeException(e);
+                                        try
+                                        {
+                                            final
+                                            ProtocolOutputConverter outputConverter =
+                                                    _session.getProtocolOutputConverter();
+
+                                            outputConverter.writeReturn(message.getMessagePublishInfo(),
+                                                                        message.getContentHeaderBody(),
+                                                                        message,
+                                                                        _channelId,
+                                                                        AMQConstant.NO_CONSUMERS.getCode(),
+                                                                        IMMEDIATE_DELIVERY_REPLY_TEXT);
+                                        }
+                                        catch (AMQException e)
+                                        {
+                                            throw new RuntimeException(e);
+                                        }
                                     }
-                                }
 
-                                @Override
-                                public void onRollback()
-                                {
+                                    @Override
+                                    public void onRollback()
+                                    {
 
+                                    }
                                 }
-                            }
-                           );
-                txn.commit();
-                ref.release();
+                               );
+                    txn.commit();
+                }
+                finally
+                {
+                    ref.release();
+                }
 
 
             }