QPID-5551 : Address review comments from Robbie Gemmell

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

3 files changed, 36 insertions, 18 deletions

diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/ExchangeRecoverer.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/ExchangeRecoverer.java
index 9857e7da72..b800734231 100644
--- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/ExchangeRecoverer.java
+++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/ExchangeRecoverer.java
@@ -66,7 +66,6 @@ public class ExchangeRecoverer extends AbstractDurableConfiguredObjectRecoverer<
                                   final Map<String, Object> attributeMap)
         {
             String exchangeName = (String) attributeMap.get(org.apache.qpid.server.model.Exchange.NAME);
-            String lifeTimePolicy = (String) attributeMap.get(org.apache.qpid.server.model.Exchange.LIFETIME_POLICY);
             try
             {
                 _exchange = _exchangeRegistry.getExchange(id);
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
index bd2dc9413c..6d4a689363 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
@@ -319,12 +319,31 @@ public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSessi
 
     private void receivedComplete()
     {
+        RuntimeException exception = null;
+
         for (AMQChannel<AMQProtocolEngine> channel : _channelsForCurrentMessage)
         {
-            channel.receivedComplete();
+            try
+            {
+                channel.receivedComplete();
+            }
+            catch(RuntimeException exceptionForThisChannel)
+            {
+                if(exception == null)
+                {
+                    exception = exceptionForThisChannel;
+                }
+                _logger.error("Error informing channel that receiving is complete. Channel: " + channel,
+                              exceptionForThisChannel);
+            }
         }
 
         _channelsForCurrentMessage.clear();
+
+        if(exception != null)
+        {
+            throw exception;
+        }
     }
 
     /**
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/BasicConsumeMethodHandler.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/BasicConsumeMethodHandler.java
index aeb5f2d2b1..cb4f758fb7 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/BasicConsumeMethodHandler.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/BasicConsumeMethodHandler.java
@@ -155,31 +155,31 @@ public class BasicConsumeMethodHandler implements StateAwareMethodListener<Basic
                 }
                 catch (AMQQueue.ExistingExclusiveConsumer e)
                 {
-                    throw body.getChannelException(AMQConstant.ACCESS_REFUSED,
-                                                   "Cannot subscribe to queue "
-                                                   + queue.getName()
-                                                   + " as it already has an existing exclusive consumer");
+                    throw body.getConnectionException(AMQConstant.ACCESS_REFUSED,
+                                                      "Cannot subscribe to queue "
+                                                      + queue.getName()
+                                                      + " as it already has an existing exclusive consumer");
                 }
                 catch (AMQQueue.ExistingConsumerPreventsExclusive e)
                 {
-                    throw body.getChannelException(AMQConstant.ACCESS_REFUSED,
-                                                   "Cannot subscribe to queue "
-                                                   + queue.getName()
-                                                   + " exclusively as it already has a consumer");
+                    throw body.getConnectionException(AMQConstant.ACCESS_REFUSED,
+                                                      "Cannot subscribe to queue "
+                                                      + queue.getName()
+                                                      + " exclusively as it already has a consumer");
                 }
                 catch (AccessControlException e)
                 {
-                    throw body.getChannelException(AMQConstant.ACCESS_REFUSED,
-                                                   "Cannot subscribe to queue "
-                                                   + queue.getName()
-                                                   + " permission denied");
+                    throw body.getConnectionException(AMQConstant.ACCESS_REFUSED,
+                                                      "Cannot subscribe to queue "
+                                                      + queue.getName()
+                                                      + " permission denied");
                 }
                 catch (MessageSource.ConsumerAccessRefused consumerAccessRefused)
                 {
-                    throw body.getChannelException(AMQConstant.ACCESS_REFUSED,
-                                                   "Cannot subscribe to queue "
-                                                   + queue.getName()
-                                                   + " as it already has an incompatible exclusivity policy");
+                    throw body.getConnectionException(AMQConstant.ACCESS_REFUSED,
+                                                      "Cannot subscribe to queue "
+                                                      + queue.getName()
+                                                      + " as it already has an incompatible exclusivity policy");
                 }
 
             }