QPID-5551 : Remove uses of AMQException, add ServerScopedRuntimeException and ConnectionScopedRuntimeException

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

15 files changed, 148 insertions, 234 deletions

diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
index 4eeb9d8fb2..b15b3f0bfa 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
@@ -29,6 +29,7 @@ import java.util.concurrent.locks.Lock;
 import org.apache.log4j.Logger;
 import org.apache.qpid.AMQConnectionException;
 import org.apache.qpid.AMQException;
+import org.apache.qpid.server.filter.AMQInvalidArgumentException;
 import org.apache.qpid.server.security.QpidSecurityException;
 import org.apache.qpid.common.AMQPFilterTypes;
 import org.apache.qpid.framing.AMQMethodBody;
@@ -81,6 +82,7 @@ import org.apache.qpid.server.txn.LocalTransaction;
 import org.apache.qpid.server.txn.LocalTransaction.ActivityTimeAccessor;
 import org.apache.qpid.server.txn.ServerTransaction;
 import org.apache.qpid.server.util.Action;
+import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 import org.apache.qpid.transport.TransportException;
 
@@ -186,9 +188,16 @@ public class AMQChannel implements AMQSessionModel, AsyncAutoCommitTransaction.F
         _transactionTimeoutHelper = new TransactionTimeoutHelper(_logSubject, new CloseAction()
         {
             @Override
-            public void doTimeoutAction(String reason) throws AMQException
+            public void doTimeoutAction(String reason)
             {
-                closeConnection(reason);
+                try
+                {
+                    closeConnection(reason);
+                }
+                catch (AMQException e)
+                {
+                    throw new ConnectionScopedRuntimeException(e);
+                }
             }
         });
     }
@@ -516,7 +525,8 @@ public class AMQChannel implements AMQSessionModel, AsyncAutoCommitTransaction.F
      */
     public AMQShortString consumeFromSource(AMQShortString tag, MessageSource source, boolean acks,
                                             FieldTable filters, boolean exclusive, boolean noLocal)
-            throws AMQException, QpidSecurityException
+            throws AMQException, QpidSecurityException, MessageSource.ExistingConsumerPreventsExclusive,
+                   MessageSource.ExistingExclusiveConsumer, AMQInvalidArgumentException
     {
         if (tag == null)
         {
@@ -579,17 +589,22 @@ public class AMQChannel implements AMQSessionModel, AsyncAutoCommitTransaction.F
                                       AMQShortString.toString(tag),
                                       options);
         }
-        catch (AMQException e)
+        catch (QpidSecurityException e)
         {
             _tag2SubscriptionTargetMap.remove(tag);
             throw e;
         }
-        catch (RuntimeException e)
+        catch (MessageSource.ExistingExclusiveConsumer e)
         {
             _tag2SubscriptionTargetMap.remove(tag);
             throw e;
         }
-        catch (QpidSecurityException e)
+        catch (MessageSource.ExistingConsumerPreventsExclusive e)
+        {
+            _tag2SubscriptionTargetMap.remove(tag);
+            throw e;
+        }
+        catch (AMQInvalidArgumentException e)
         {
             _tag2SubscriptionTargetMap.remove(tag);
             throw e;
@@ -601,9 +616,8 @@ public class AMQChannel implements AMQSessionModel, AsyncAutoCommitTransaction.F
      * Unsubscribe a consumer from a queue.
      * @param consumerTag
      * @return true if the consumerTag had a mapped queue that could be unregistered.
-     * @throws AMQException
      */
-    public boolean unsubscribeConsumer(AMQShortString consumerTag) throws AMQException
+    public boolean unsubscribeConsumer(AMQShortString consumerTag)
     {
 
         ConsumerTarget_0_8 target = _tag2SubscriptionTargetMap.remove(consumerTag);
@@ -622,16 +636,14 @@ public class AMQChannel implements AMQSessionModel, AsyncAutoCommitTransaction.F
 
     /**
      * Called from the protocol session to close this channel and clean up. T
-     *
-     * @throws AMQException if there is an error during closure
      */
     @Override
-    public void close() throws AMQException
+    public void close()
     {
         close(null, null);
     }
 
-    public void close(AMQConstant cause, String message) throws AMQException
+    public void close(AMQConstant cause, String message)
     {
         if(!_closing.compareAndSet(false, true))
         {
@@ -651,17 +663,13 @@ public class AMQChannel implements AMQSessionModel, AsyncAutoCommitTransaction.F
         {
             requeue();
         }
-        catch (AMQException e)
-        {
-            _logger.error("Caught AMQException whilst attempting to requeue:" + e);
-        }
         catch (TransportException e)
         {
             _logger.error("Caught TransportException whilst attempting to requeue:" + e);
         }
     }
 
-    private void unsubscribeAllConsumers() throws AMQException
+    private void unsubscribeAllConsumers()
     {
         if (_logger.isInfoEnabled())
         {
@@ -724,9 +732,8 @@ public class AMQChannel implements AMQSessionModel, AsyncAutoCommitTransaction.F
      * Called to attempt re-delivery all outstanding unacknowledged messages on the channel. May result in delivery to
      * this same channel or to other subscribers.
      *
-     * @throws org.apache.qpid.AMQException if the requeue fails
      */
-    public void requeue() throws AMQException
+    public void requeue()
     {
         // we must create a new map since all the messages will get a new delivery tag when they are redelivered
         Collection<MessageInstance> messagesToBeDelivered = _unacknowledgedMessageMap.cancelAllMessages();
@@ -756,9 +763,8 @@ public class AMQChannel implements AMQSessionModel, AsyncAutoCommitTransaction.F
      *
      * @param deliveryTag The message to requeue
      *
-     * @throws AMQException If something goes wrong.
      */
-    public void requeue(long deliveryTag) throws AMQException
+    public void requeue(long deliveryTag)
     {
         MessageInstance unacked = _unacknowledgedMessageMap.remove(deliveryTag);
 
@@ -1455,7 +1461,7 @@ public class AMQChannel implements AMQSessionModel, AsyncAutoCommitTransaction.F
         return getProtocolSession().getVirtualHost();
     }
 
-    public void checkTransactionStatus(long openWarn, long openClose, long idleWarn, long idleClose) throws AMQException
+    public void checkTransactionStatus(long openWarn, long openClose, long idleWarn, long idleClose)
     {
         _transactionTimeoutHelper.checkIdleOrOpenTimes(_transaction, openWarn, openClose, idleWarn, idleClose);
     }
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 329aa396b0..5e95701e5a 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
@@ -46,28 +46,11 @@ import org.apache.log4j.Logger;
 import org.apache.qpid.AMQChannelException;
 import org.apache.qpid.AMQConnectionException;
 import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.*;
 import org.apache.qpid.server.security.QpidSecurityException;
 import org.apache.qpid.codec.AMQCodecFactory;
 import org.apache.qpid.common.QpidProperties;
 import org.apache.qpid.common.ServerPropertyNames;
-import org.apache.qpid.framing.AMQBody;
-import org.apache.qpid.framing.AMQDataBlock;
-import org.apache.qpid.framing.AMQFrame;
-import org.apache.qpid.framing.AMQMethodBody;
-import org.apache.qpid.framing.AMQProtocolHeaderException;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.ChannelCloseBody;
-import org.apache.qpid.framing.ChannelCloseOkBody;
-import org.apache.qpid.framing.ConnectionCloseBody;
-import org.apache.qpid.framing.ContentBody;
-import org.apache.qpid.framing.ContentHeaderBody;
-import org.apache.qpid.framing.FieldTable;
-import org.apache.qpid.framing.FieldTableFactory;
-import org.apache.qpid.framing.HeartbeatBody;
-import org.apache.qpid.framing.MethodDispatcher;
-import org.apache.qpid.framing.MethodRegistry;
-import org.apache.qpid.framing.ProtocolInitiation;
-import org.apache.qpid.framing.ProtocolVersion;
 import org.apache.qpid.properties.ConnectionStartProperties;
 import org.apache.qpid.protocol.AMQConstant;
 import org.apache.qpid.protocol.AMQMethodEvent;
@@ -95,6 +78,8 @@ import org.apache.qpid.server.protocol.v0_8.state.AMQState;
 import org.apache.qpid.server.protocol.v0_8.state.AMQStateManager;
 import org.apache.qpid.server.stats.StatisticsCounter;
 import org.apache.qpid.server.consumer.Consumer;
+import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
+import org.apache.qpid.server.util.ServerScopedRuntimeException;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 import org.apache.qpid.transport.Sender;
 import org.apache.qpid.transport.TransportException;
@@ -303,9 +288,24 @@ public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSessi
             }
             receivedComplete();
         }
-        catch (Exception e)
+        catch (ConnectionScopedRuntimeException e)
+        {
+            _logger.error("Unexpected exception", e);
+            closeProtocolSession();
+        }
+        catch (AMQProtocolVersionException e)
+        {
+            _logger.error("Unexpected protocol version", e);
+            closeProtocolSession();
+        }
+        catch (AMQFrameDecodingException e)
         {
-            _logger.error("Unexpected exception when processing datablocks", e);
+            _logger.error("Frame decoding", e);
+            closeProtocolSession();
+        }
+        catch (IOException e)
+        {
+            _logger.error("I/O Exception", e);
             closeProtocolSession();
         }
         finally
@@ -314,34 +314,14 @@ public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSessi
         }
     }
 
-    private void receivedComplete() throws AMQException
+    private void receivedComplete()
     {
-        Exception exception = null;
         for (AMQChannel channel : _channelsForCurrentMessage)
         {
-            try
-            {
-                channel.receivedComplete();
-            }
-            catch(Exception exceptionForThisChannel)
-            {
-                if(exception == null)
-                {
-                    exception = exceptionForThisChannel;
-                }
-                _logger.error("Error informing channel that receiving is complete. Channel: " + channel, exceptionForThisChannel);
-            }
+            channel.receivedComplete();
         }
 
         _channelsForCurrentMessage.clear();
-
-        if(exception != null)
-        {
-            throw new AMQException(
-                    AMQConstant.INTERNAL_ERROR,
-                    "Error informing channel that receiving is complete: " + exception.getMessage(),
-                    exception);
-        }
     }
 
     /**
@@ -549,7 +529,7 @@ public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSessi
         }
         catch (IOException e)
         {
-            throw new RuntimeException(e);
+            throw new ServerScopedRuntimeException(e);
         }
 
         final ByteBuffer buf;
@@ -812,16 +792,15 @@ public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSessi
      *
      * @param channelId id of the channel to close
      *
-     * @throws AMQException             if an error occurs closing the channel
      * @throws IllegalArgumentException if the channel id is not valid
      */
     @Override
-    public void closeChannel(int channelId) throws AMQException
+    public void closeChannel(int channelId)
     {
         closeChannel(channelId, null, null);
     }
 
-    public void closeChannel(int channelId, AMQConstant cause, String message) throws AMQException
+    public void closeChannel(int channelId, AMQConstant cause, String message)
     {
         final AMQChannel channel = getChannel(channelId);
         if (channel == null)
@@ -903,7 +882,7 @@ public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSessi
      *
      * @throws AMQException if an error occurs while closing any channel
      */
-    private void closeAllChannels() throws AMQException
+    private void closeAllChannels()
     {
         for (AMQChannel channel : getChannels())
         {
@@ -921,7 +900,7 @@ public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSessi
 
     /** This must be called when the session is _closed in order to free up any resources managed by the session. */
     @Override
-    public void closeSession() throws AMQException
+    public void closeSession()
     {
         if(_closing.compareAndSet(false,true))
         {
@@ -996,7 +975,7 @@ public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSessi
         }
     }
 
-    private void closeConnection(int channelId, AMQConnectionException e) throws AMQException
+    private void closeConnection(int channelId, AMQConnectionException e)
     {
         try
         {
@@ -1033,7 +1012,7 @@ public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSessi
         {
             _stateManager.changeState(AMQState.CONNECTION_CLOSED);
         }
-        catch (AMQException e)
+        catch (ConnectionScopedRuntimeException e)
         {
             _logger.info(e.getMessage());
         }
@@ -1234,9 +1213,9 @@ public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSessi
                 closeProtocolSession();
             }
         }
-        catch (AMQException e)
+        catch (ConnectionScopedRuntimeException e)
         {
-           _logger.error("Could not close protocol engine", e);
+            _logger.error("Could not close protocol engine", e);
         }
         catch (TransportException e)
         {
@@ -1269,15 +1248,30 @@ public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSessi
         }
         else
         {
-            _logger.error("Exception caught in " + this + ", closing connection explicitly: " + throwable, throwable);
+            try
+            {
+                _logger.error("Exception caught in " + this + ", closing connection explicitly: " + throwable, throwable);
 
 
-            MethodRegistry methodRegistry = MethodRegistry.getMethodRegistry(getProtocolVersion());
-            ConnectionCloseBody closeBody = methodRegistry.createConnectionCloseBody(200,new AMQShortString(throwable.getMessage()),0,0);
+                MethodRegistry methodRegistry = MethodRegistry.getMethodRegistry(getProtocolVersion());
+                ConnectionCloseBody closeBody = methodRegistry.createConnectionCloseBody(200,new AMQShortString(throwable.getMessage()),0,0);
 
-            writeFrame(closeBody.generateFrame(0));
+                writeFrame(closeBody.generateFrame(0));
 
-            _sender.close();
+                _sender.close();
+            }
+            finally
+            {
+                if(throwable instanceof Error)
+                {
+                    throw (Error) throwable;
+                }
+                if(throwable instanceof ServerScopedRuntimeException)
+                {
+                    throw (ServerScopedRuntimeException) throwable;
+                }
+
+            }
         }
     }
 
@@ -1441,15 +1435,8 @@ public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSessi
         {
             writeFrame(responseBody.generateFrame(0));
 
-            try
-            {
+            closeSession();
 
-                closeSession();
-            }
-            catch (AMQException ex)
-            {
-                throw new RuntimeException(ex);
-            }
         }
         finally
         {
@@ -1483,15 +1470,7 @@ public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSessi
         try
         {
             writeFrame(responseBody.generateFrame(channelId));
-
-            try
-            {
-                closeChannel(channelId);
-            }
-            catch (AMQException ex)
-            {
-                throw new RuntimeException(ex);
-            }
+            closeChannel(channelId);
         }
         finally
         {
@@ -1507,7 +1486,7 @@ public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSessi
         return getContextKey().toString();
     }
 
-    public void closeSession(AMQSessionModel session, AMQConstant cause, String message) throws AMQException
+    public void closeSession(AMQSessionModel session, AMQConstant cause, String message)
     {
         int channelId = ((AMQChannel)session).getChannelId();
         closeChannel(channelId, cause, message);
@@ -1522,7 +1501,7 @@ public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSessi
         writeFrame(responseBody.generateFrame(channelId));
     }
 
-    public void close(AMQConstant cause, String message) throws AMQException
+    public void close(AMQConstant cause, String message)
     {
         closeConnection(0, new AMQConnectionException(cause, message, 0, 0,
 		                getProtocolOutputConverter().getProtocolMajorVersion(),
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolSession.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolSession.java
index 6bcd4b9d49..58a3b5df12 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolSession.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolSession.java
@@ -71,7 +71,7 @@ public interface AMQProtocolSession extends AMQVersionAwareProtocolSession, Auth
 
     public static interface Task
     {
-        public void doTask(AMQProtocolSession session) throws AMQException;
+        public void doTask(AMQProtocolSession session);
     }
 
     /**
@@ -152,7 +152,7 @@ public interface AMQProtocolSession extends AMQVersionAwareProtocolSession, Auth
     void initHeartbeats(int delay);
 
     /** This must be called when the session is _closed in order to free up any resources managed by the session. */
-    void closeSession() throws AMQException;
+    void closeSession();
 
     void closeProtocolSession();
 
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_Internal_to_v0_8.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_Internal_to_v0_8.java
index b80ad3e7b8..f2bb95c8d5 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_Internal_to_v0_8.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_Internal_to_v0_8.java
@@ -29,6 +29,7 @@ import org.apache.qpid.server.message.internal.InternalMessage;
 import org.apache.qpid.server.plugin.MessageConverter;
 import org.apache.qpid.server.store.StoreFuture;
 import org.apache.qpid.server.store.StoredMessage;
+import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 import org.apache.qpid.transport.codec.BBEncoder;
 
@@ -236,7 +237,7 @@ public class MessageConverter_Internal_to_v0_8 implements MessageConverter<Inter
             }
             catch (IOException e)
             {
-                throw new RuntimeException(e);
+                throw new ConnectionScopedRuntimeException(e);
             }
         }
     }
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_v0_8_to_Internal.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_v0_8_to_Internal.java
index 6076ff66c7..f35d37ecbd 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_v0_8_to_Internal.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_v0_8_to_Internal.java
@@ -22,6 +22,7 @@ package org.apache.qpid.server.protocol.v0_8;
 
 import org.apache.qpid.server.message.internal.InternalMessage;
 import org.apache.qpid.server.plugin.MessageConverter;
+import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 import org.apache.qpid.transport.codec.BBDecoder;
 import org.apache.qpid.typedmessage.TypedBytesContentReader;
@@ -124,11 +125,11 @@ public class MessageConverter_v0_8_to_Internal implements MessageConverter<AMQMe
                 }
                 catch (TypedBytesFormatException e)
                 {
-                    throw new RuntimeException(e);  // TODO - Implement
+                    throw new ConnectionScopedRuntimeException(e);
                 }
                 catch (EOFException e)
                 {
-                    throw new RuntimeException(e);  // TODO - Implement
+                    throw new ConnectionScopedRuntimeException(e);
                 }
             }
             return list;
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageMetaData.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageMetaData.java
index 3665e7f135..fd7fb9ca80 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageMetaData.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageMetaData.java
@@ -22,6 +22,8 @@ package org.apache.qpid.server.protocol.v0_8;
 
 import java.util.Collection;
 import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQFrameDecodingException;
+import org.apache.qpid.framing.AMQProtocolVersionException;
 import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.framing.BasicContentHeaderProperties;
 import org.apache.qpid.framing.ContentHeaderBody;
@@ -32,6 +34,7 @@ import org.apache.qpid.server.message.AMQMessageHeader;
 import org.apache.qpid.server.plugin.MessageMetaDataType;
 import org.apache.qpid.server.store.StorableMessageMetaData;
 import org.apache.qpid.server.util.ByteBufferOutputStream;
+import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
 import org.apache.qpid.util.ByteBufferInputStream;
 
 import java.io.DataInputStream;
@@ -132,7 +135,7 @@ public class MessageMetaData implements StorableMessageMetaData
         catch (IOException e)
         {
             // This shouldn't happen as we are not actually using anything that can throw an IO Exception
-            throw new RuntimeException(e);
+            throw new ConnectionScopedRuntimeException(e);
         }
 
         return dest.position()-oldPosition;
@@ -196,17 +199,21 @@ public class MessageMetaData implements StorableMessageMetaData
                         };
                 return new MessageMetaData(publishBody, chb, arrivalTime);
             }
-            catch (AMQException e)
+            catch (IOException e)
             {
-                throw new RuntimeException(e);
+                throw new ConnectionScopedRuntimeException(e);
             }
-            catch (IOException e)
+            catch (AMQProtocolVersionException e)
             {
-                throw new RuntimeException(e);
+                throw new ConnectionScopedRuntimeException(e);
+            }
+            catch (AMQFrameDecodingException e)
+            {
+                throw new ConnectionScopedRuntimeException(e);
             }
 
         }
-    };
+    }
 
     public AMQMessageHeader getMessageHeader()
     {
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 ad4235b786..ce90de7aac 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
@@ -28,6 +28,7 @@ import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.framing.BasicConsumeBody;
 import org.apache.qpid.framing.MethodRegistry;
 import org.apache.qpid.protocol.AMQConstant;
+import org.apache.qpid.server.filter.AMQInvalidArgumentException;
 import org.apache.qpid.server.message.MessageSource;
 import org.apache.qpid.server.protocol.v0_8.AMQChannel;
 import org.apache.qpid.server.protocol.v0_8.AMQProtocolSession;
@@ -149,7 +150,7 @@ public class BasicConsumeMethodHandler implements StateAwareMethodListener<Basic
                     }
 
                 }
-                catch (org.apache.qpid.AMQInvalidArgumentException ise)
+                catch (AMQInvalidArgumentException ise)
                 {
                     _logger.debug("Closing connection due to invalid selector");
 
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/BasicGetMethodHandler.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/BasicGetMethodHandler.java
index f55a120a2d..c9a7cc69a1 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/BasicGetMethodHandler.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/BasicGetMethodHandler.java
@@ -31,6 +31,7 @@ import org.apache.qpid.framing.MethodRegistry;
 import org.apache.qpid.protocol.AMQConstant;
 import org.apache.qpid.server.message.InstanceProperties;
 import org.apache.qpid.server.message.MessageInstance;
+import org.apache.qpid.server.message.MessageSource;
 import org.apache.qpid.server.message.ServerMessage;
 import org.apache.qpid.server.protocol.v0_8.AMQChannel;
 import org.apache.qpid.server.flow.FlowCreditManager;
@@ -124,6 +125,17 @@ public class BasicGetMethodHandler implements StateAwareMethodListener<BasicGetB
                     throw body.getConnectionException(AMQConstant.ACCESS_REFUSED,
                                                       e.getMessage());
                 }
+                catch (MessageSource.ExistingExclusiveConsumer e)
+                {
+                    throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
+                                                      "Queue has an exclusive consumer");
+                }
+                catch (MessageSource.ExistingConsumerPreventsExclusive e)
+                {
+                    throw body.getConnectionException(AMQConstant.INTERNAL_ERROR,
+                                                      "The GET request has been evaluated as an exclusive consumer, " +
+                                                      "this is likely due to a programming error in the Qpid broker");
+                }
             }
         }
     }
@@ -132,7 +144,8 @@ public class BasicGetMethodHandler implements StateAwareMethodListener<BasicGetB
                                      final AMQProtocolSession session,
                                      final AMQChannel channel,
                                      final boolean acks)
-            throws AMQException, QpidSecurityException
+            throws AMQException, QpidSecurityException, MessageSource.ExistingConsumerPreventsExclusive,
+                   MessageSource.ExistingExclusiveConsumer
     {
 
         final FlowCreditManager singleMessageCredit = new MessageOnlyCreditManager(1L);
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/ChannelOpenHandler.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/ChannelOpenHandler.java
index 442c912032..2594242db4 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/ChannelOpenHandler.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/ChannelOpenHandler.java
@@ -35,6 +35,7 @@ import org.apache.qpid.server.protocol.v0_8.AMQChannel;
 import org.apache.qpid.server.protocol.v0_8.AMQProtocolSession;
 import org.apache.qpid.server.protocol.v0_8.state.AMQStateManager;
 import org.apache.qpid.server.protocol.v0_8.state.StateAwareMethodListener;
+import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 
 import java.io.ByteArrayOutputStream;
@@ -99,7 +100,7 @@ public class ChannelOpenHandler implements StateAwareMethodListener<ChannelOpenB
             catch (IOException e)
             {
                 // This *really* shouldn't happen as we're not doing any I/O
-                throw new RuntimeException("I/O exception when writing to byte array", e);
+                throw new ConnectionScopedRuntimeException("I/O exception when writing to byte array", e);
             }
 
             // should really associate this channelId to the session
@@ -123,7 +124,7 @@ public class ChannelOpenHandler implements StateAwareMethodListener<ChannelOpenB
             catch (IOException e)
             {
                 // This *really* shouldn't happen as we're not doing any I/O
-                throw new RuntimeException("I/O exception when writing to byte array", e);
+                throw new ConnectionScopedRuntimeException("I/O exception when writing to byte array", e);
             }
 
             // should really associate this channelId to the session
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/ExchangeDeclareHandler.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/ExchangeDeclareHandler.java
index fa513486a4..5c5b1f141b 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/ExchangeDeclareHandler.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/ExchangeDeclareHandler.java
@@ -24,7 +24,7 @@ import org.apache.log4j.Logger;
 
 import org.apache.qpid.AMQConnectionException;
 import org.apache.qpid.AMQException;
-import org.apache.qpid.AMQUnknownExchangeType;
+import org.apache.qpid.server.exchange.AMQUnknownExchangeType;
 import org.apache.qpid.framing.AMQMethodBody;
 import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.framing.ExchangeDeclareBody;
@@ -38,6 +38,7 @@ import org.apache.qpid.server.protocol.v0_8.state.StateAwareMethodListener;
 import org.apache.qpid.server.security.QpidSecurityException;
 import org.apache.qpid.server.virtualhost.ExchangeExistsException;
 import org.apache.qpid.server.virtualhost.ReservedExchangeNameException;
+import org.apache.qpid.server.virtualhost.UnknownExchangeException;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 
 public class ExchangeDeclareHandler implements StateAwareMethodListener<ExchangeDeclareBody>
@@ -129,6 +130,11 @@ public class ExchangeDeclareHandler implements StateAwareMethodListener<Exchange
             {
                 throw body.getConnectionException(AMQConstant.ACCESS_REFUSED, e.getMessage());
             }
+            catch (UnknownExchangeException e)
+            {
+                // note - since 0-8/9/9-1 can't set the alt. exchange this exception should never occur
+                throw body.getConnectionException(AMQConstant.NOT_FOUND, "Unknown alternate exchange",e);
+            }
         }
 
 
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/QueueDeclareHandler.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/QueueDeclareHandler.java
index 1286a20970..5b5525643c 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/QueueDeclareHandler.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/QueueDeclareHandler.java
@@ -34,21 +34,17 @@ import org.apache.qpid.server.model.UUIDGenerator;
 import org.apache.qpid.server.protocol.v0_8.AMQProtocolSession;
 import org.apache.qpid.server.protocol.AMQSessionModel;
 import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.queue.AMQQueueFactory;
 import org.apache.qpid.server.queue.QueueArgumentsConverter;
-import org.apache.qpid.server.queue.QueueRegistry;
 import org.apache.qpid.server.protocol.v0_8.state.AMQStateManager;
 import org.apache.qpid.server.protocol.v0_8.state.StateAwareMethodListener;
 import org.apache.qpid.server.security.QpidSecurityException;
-import org.apache.qpid.server.store.DurableConfigurationStoreHelper;
-import org.apache.qpid.server.store.DurableConfigurationStore;
 import org.apache.qpid.server.util.Action;
 import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 
 import java.util.Map;
 import java.util.UUID;
-import org.apache.qpid.server.virtualhost.plugins.QueueExistsException;
+import org.apache.qpid.server.virtualhost.QueueExistsException;
 
 public class QueueDeclareHandler implements StateAwareMethodListener<QueueDeclareBody>
 {
@@ -131,7 +127,7 @@ public class QueueDeclareHandler implements StateAwareMethodListener<QueueDeclar
                         final AMQQueue q = queue;
                         final AMQProtocolSession.Task sessionCloseTask = new AMQProtocolSession.Task()
                         {
-                            public void doTask(AMQProtocolSession session) throws AMQException
+                            public void doTask(AMQProtocolSession session)
                             {
                                 q.setExclusiveOwningSession(null);
                             }
@@ -219,7 +215,7 @@ public class QueueDeclareHandler implements StateAwareMethodListener<QueueDeclar
                                    QueueDeclareBody body,
                                    final VirtualHost virtualHost,
                                    final AMQProtocolSession session)
-            throws AMQException, QpidSecurityException
+            throws AMQException, QpidSecurityException, QueueExistsException
     {
 
         final boolean durable = body.getDurable();
@@ -241,7 +237,7 @@ public class QueueDeclareHandler implements StateAwareMethodListener<QueueDeclar
             final AMQProtocolSession.Task deleteQueueTask =
                     new AMQProtocolSession.Task()
                     {
-                        public void doTask(AMQProtocolSession session) throws AMQException
+                        public void doTask(AMQProtocolSession session)
                         {
                             if (virtualHost.getQueue(queueName.toString()) == queue)
                             {
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/state/AMQStateManager.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/state/AMQStateManager.java
index 0555bba98b..c3cbb5e26f 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/state/AMQStateManager.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/state/AMQStateManager.java
@@ -52,8 +52,6 @@ public class AMQStateManager implements AMQMethodListener
     /** The current state */
     private AMQState _currentState;
 
-    private CopyOnWriteArraySet<StateListener> _stateListeners = new CopyOnWriteArraySet<StateListener>();
-
     public AMQStateManager(Broker broker, AMQProtocolSession protocolSession)
     {
         _broker = broker;
@@ -72,30 +70,17 @@ public class AMQStateManager implements AMQMethodListener
         return _broker;
     }
 
-    public AMQState getCurrentState()
-    {
-        return _currentState;
-    }
-
-    public void changeState(AMQState newState) throws AMQException
+    public void changeState(AMQState newState)
     {
         _logger.debug("State changing to " + newState + " from old state " + _currentState);
         final AMQState oldState = _currentState;
         _currentState = newState;
 
-        for (StateListener l : _stateListeners)
-        {
-            l.stateChanged(oldState, newState);
-        }
     }
 
     public void error(Exception e)
     {
         _logger.error("State manager received error notification[Current State:" + _currentState + "]: " + e, e);
-        for (StateListener l : _stateListeners)
-        {
-            l.error(e);
-        }
     }
 
     public <B extends AMQMethodBody> boolean methodReceived(AMQMethodEvent<B> evt) throws AMQException
@@ -121,28 +106,6 @@ public class AMQStateManager implements AMQMethodListener
 
     }
 
-    private <B extends AMQMethodBody> void checkChannel(AMQMethodEvent<B> evt, AMQProtocolSession protocolSession)
-        throws AMQException
-    {
-        if ((evt.getChannelId() != 0) && !(evt.getMethod() instanceof ChannelOpenBody)
-                && (protocolSession.getChannel(evt.getChannelId()) == null)
-                && !protocolSession.channelAwaitingClosure(evt.getChannelId()))
-        {
-            throw evt.getMethod().getChannelNotFoundException(evt.getChannelId());
-        }
-    }
-
-    public void addStateListener(StateListener listener)
-    {
-        _logger.debug("Adding state listener");
-        _stateListeners.add(listener);
-    }
-
-    public void removeStateListener(StateListener listener)
-    {
-        _stateListeners.remove(listener);
-    }
-
     public VirtualHostRegistry getVirtualHostRegistry()
     {
         return _broker.getVirtualHostRegistry();
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/state/IllegalStateTransitionException.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/state/IllegalStateTransitionException.java
deleted file mode 100644
index f61553f8a2..0000000000
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/state/IllegalStateTransitionException.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-package org.apache.qpid.server.protocol.v0_8.state;
-
-import org.apache.qpid.AMQException;
-
-/**
- * @todo Not an AMQP exception as no status code.
- *
- * @todo Not used! Delete.
- */
-public class IllegalStateTransitionException extends AMQException
-{
-    private AMQState _originalState;
-
-    private Class _frame;
-
-    public IllegalStateTransitionException(AMQState originalState, Class frame)
-    {
-        super("No valid state transition defined for receiving frame " + frame + " from state " + originalState);
-        _originalState = originalState;
-        _frame = frame;
-    }
-
-    public AMQState getOriginalState()
-    {
-        return _originalState;
-    }
-
-    public Class getFrameClass()
-    {
-        return _frame;
-    }
-}
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AckTest.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AckTest.java
index 23c32f988d..8d7de4cd93 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AckTest.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AckTest.java
@@ -139,28 +139,20 @@ public class AckTest extends QpidTestCase
             final StoredMessage storedMessage = _messageStore.addMessage(mmd);
             final AMQMessage message = new AMQMessage(storedMessage);
             ServerTransaction txn = new AutoCommitTransaction(_messageStore);
-            txn.enqueue(_queue, message, new ServerTransaction.Action() {
-                public void postCommit()
-                {
-                    try
-                    {
-
-                        _queue.enqueue(message,null);
-                    }
-                    catch (AMQException e)
-                    {
-                         throw new RuntimeException(e);
-                    }
-                }
-
-                public void onRollback()
-                {
-                    //To change body of implemented methods use File | Settings | File Templates.
-                }
-            });
+            txn.enqueue(_queue, message,
+                        new ServerTransaction.Action()
+                        {
+                            public void postCommit()
+                            {
+                                _queue.enqueue(message,null);
+                            }
+
+                            public void onRollback()
+                            {
+                                //To change body of implemented methods use File | Settings | File Templates.
+                            }
+                        });
 
-            // we manually send the message to the subscription
-            //_subscription.send(new QueueEntry(_queue,msg), _queue);
         }
         try
         {
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/InternalTestProtocolSession.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/InternalTestProtocolSession.java
index 4a2fbe44fa..7661d98cb4 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/InternalTestProtocolSession.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/InternalTestProtocolSession.java
@@ -223,7 +223,7 @@ public class InternalTestProtocolSession extends AMQProtocolEngine implements Pr
         //  Then the AMQMinaProtocolSession can join on the returning future without a NPE.
     }
 
-    public void closeSession(AMQSessionModel session, AMQConstant cause, String message) throws AMQException
+    public void closeSession(AMQSessionModel session, AMQConstant cause, String message)
     {
         super.closeSession(session, cause, message);