QPID-567 : Add mutliversion support to Qpid/Java, fixed client support when server returns Protocol header.

           Added QueueUnbind
           Added ability to select protocol version in ConnectionURL or with -Dorg.apache.qpid.amqp_version
           

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/M2.1@598105 13f79535-47bb-0310-9956-ffa450edef68

6 files changed, 128 insertions, 6 deletions

diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/DestNameExchange.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/DestNameExchange.java
index 5edffc19ed..1b5bf4f343 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/exchange/DestNameExchange.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/DestNameExchange.java
@@ -34,6 +34,7 @@ import javax.management.openmbean.TabularDataSupport;
 
 import org.apache.log4j.Logger;
 import org.apache.qpid.AMQException;
+import org.apache.qpid.protocol.AMQConstant;
 import org.apache.qpid.exchange.ExchangeDefaults;
 import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.framing.FieldTable;
@@ -150,7 +151,7 @@ public class DestNameExchange extends AbstractExchange
 
         if (!_index.remove(routingKey, queue))
         {
-            throw new AMQException("Queue " + queue + " was not registered with exchange " + this.getName() +
+            throw new AMQException(AMQConstant.NOT_FOUND, "Queue " + queue + " was not registered with exchange " + this.getName() +
                                    " with routing key " + routingKey + ". No queue was registered with that _routing key");
         }
     }
diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java
index b55dbcc792..3da152c528 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java
@@ -22,6 +22,7 @@ package org.apache.qpid.server.exchange;
 
 import org.apache.log4j.Logger;
 import org.apache.qpid.AMQException;
+import org.apache.qpid.protocol.AMQConstant;
 import org.apache.qpid.exchange.ExchangeDefaults;
 import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.framing.FieldTable;
@@ -277,7 +278,7 @@ public class DestWildExchange extends AbstractExchange
         List<AMQQueue> queues = _routingKey2queues.get(routingKey);
         if (queues == null)
         {
-            throw new AMQException("Queue " + queue + " was not registered with exchange " + this.getName()
+            throw new AMQException(AMQConstant.NOT_FOUND, "Queue " + queue + " was not registered with exchange " + this.getName()
                                    + " with routing key " + routingKey + ". No queue was registered with that _routing key");
 
         }
@@ -285,7 +286,7 @@ public class DestWildExchange extends AbstractExchange
         boolean removedQ = queues.remove(queue);
         if (!removedQ)
         {
-            throw new AMQException("Queue " + queue + " was not registered with exchange " + this.getName()
+            throw new AMQException(AMQConstant.NOT_FOUND, "Queue " + queue + " was not registered with exchange " + this.getName()
                                    + " with routing key " + routingKey);
         }
 
diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java
index aa13f1d8ee..82d57770df 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java
@@ -22,6 +22,7 @@ package org.apache.qpid.server.exchange;
 

 import org.apache.log4j.Logger;

 import org.apache.qpid.AMQException;

+import org.apache.qpid.protocol.AMQConstant;

 import org.apache.qpid.exchange.ExchangeDefaults;

 import org.apache.qpid.framing.AMQShortString;

 import org.apache.qpid.framing.FieldTable;

@@ -146,7 +147,7 @@ public class FanoutExchange extends AbstractExchange
 

         if (!_queues.remove(queue))

         {

-            throw new AMQException("Queue " + queue + " was not registered with exchange " + this.getName() + ". ");

+            throw new AMQException(AMQConstant.NOT_FOUND, "Queue " + queue + " was not registered with exchange " + this.getName() + ". ");

         }

     }

 

diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java
index 3544e9d1f8..31ae799877 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java
@@ -22,6 +22,7 @@ package org.apache.qpid.server.exchange;
 
 import org.apache.log4j.Logger;
 import org.apache.qpid.AMQException;
+import org.apache.qpid.protocol.AMQConstant;
 import org.apache.qpid.exchange.ExchangeDefaults;
 import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.framing.AMQTypedValue;
@@ -200,7 +201,11 @@ public class HeadersExchange extends AbstractExchange
     public void deregisterQueue(AMQShortString routingKey, AMQQueue queue, FieldTable args) throws AMQException
     {
         _logger.debug("Exchange " + getName() + ": Unbinding " + queue.getName());
-        _bindings.remove(new Registration(new HeadersBinding(args), queue));
+        if(!_bindings.remove(new Registration(new HeadersBinding(args), queue)))
+        {
+            throw new AMQException(AMQConstant.NOT_FOUND, "Queue " + queue + " was not registered with exchange " + this.getName()
+                                   + " with headers args " + args);    
+        }
     }
 
     public void route(AMQMessage payload) throws AMQException
diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/QueueUnbindHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/QueueUnbindHandler.java
new file mode 100644
index 0000000000..b056fa6797
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/handler/QueueUnbindHandler.java
@@ -0,0 +1,110 @@
+package org.apache.qpid.server.handler;

+

+import org.apache.log4j.Logger;

+

+import org.apache.qpid.framing.*;

+import org.apache.qpid.framing.amqp_0_9.MethodRegistry_0_9;

+import org.apache.qpid.server.state.StateAwareMethodListener;

+import org.apache.qpid.server.state.AMQStateManager;

+import org.apache.qpid.server.protocol.AMQProtocolSession;

+import org.apache.qpid.server.virtualhost.VirtualHost;

+import org.apache.qpid.server.exchange.ExchangeRegistry;

+import org.apache.qpid.server.exchange.Exchange;

+import org.apache.qpid.server.queue.QueueRegistry;

+import org.apache.qpid.server.queue.AMQQueue;

+import org.apache.qpid.server.AMQChannel;

+import org.apache.qpid.AMQException;

+import org.apache.qpid.AMQInvalidRoutingKeyException;

+import org.apache.qpid.protocol.AMQConstant;

+

+public class QueueUnbindHandler implements StateAwareMethodListener<QueueUnbindBody>

+{

+    private static final Logger _log = Logger.getLogger(QueueUnbindHandler.class);

+

+    private static final QueueUnbindHandler _instance = new QueueUnbindHandler();

+

+    public static QueueUnbindHandler getInstance()

+    {

+        return _instance;

+    }

+

+    private QueueUnbindHandler()

+    {

+    }

+

+    public void methodReceived(AMQStateManager stateManager, QueueUnbindBody body, int channelId) throws AMQException

+    {

+        AMQProtocolSession session = stateManager.getProtocolSession();

+        VirtualHost virtualHost = session.getVirtualHost();

+        ExchangeRegistry exchangeRegistry = virtualHost.getExchangeRegistry();

+        QueueRegistry queueRegistry = virtualHost.getQueueRegistry();

+

+

+        final AMQQueue queue;

+        final AMQShortString routingKey;

+

+        if (body.getQueue() == null)

+        {

+            AMQChannel channel = session.getChannel(channelId);

+

+            if (channel == null)

+            {

+                throw body.getChannelNotFoundException(channelId);

+            }

+

+            queue = channel.getDefaultQueue();

+

+            if (queue == null)

+            {

+                throw body.getConnectionException(AMQConstant.NOT_FOUND, "No default queue defined on channel and queue was null");

+            }

+

+            routingKey = body.getRoutingKey() == null ? null : body.getRoutingKey().intern();

+

+        }

+        else

+        {

+            queue = queueRegistry.getQueue(body.getQueue());

+            routingKey = body.getRoutingKey() == null ? null : body.getRoutingKey().intern();

+        }

+

+        if (queue == null)

+        {

+            throw body.getConnectionException(AMQConstant.NOT_FOUND, "Queue " + body.getQueue() + " does not exist.");

+        }

+        final Exchange exch = exchangeRegistry.getExchange(body.getExchange());

+        if (exch == null)

+        {

+            throw body.getChannelException(AMQConstant.NOT_FOUND, "Exchange " + body.getExchange() + " does not exist.");

+        }

+

+

+        try

+        {

+            queue.unBind(routingKey, body.getArguments(), exch);

+        }

+        catch (AMQInvalidRoutingKeyException rke)

+        {

+            throw body.getChannelException(AMQConstant.INVALID_ROUTING_KEY, routingKey.toString());

+        }

+        catch (AMQException e)

+        {

+            if(e.getErrorCode() == AMQConstant.NOT_FOUND)

+            {

+                throw body.getConnectionException(AMQConstant.NOT_FOUND,e.getMessage(),e);

+            }

+            throw body.getChannelException(AMQConstant.CHANNEL_ERROR, e.toString());

+        }

+

+        if (_log.isInfoEnabled())

+        {

+            _log.info("Binding queue " + queue + " to exchange " + exch + " with routing key " + routingKey);

+        }

+

+        MethodRegistry_0_9 methodRegistry = (MethodRegistry_0_9) session.getMethodRegistry();

+        AMQMethodBody responseBody = methodRegistry.createQueueUnbindOkBody();

+        session.writeFrame(responseBody.generateFrame(channelId));

+

+

+    }

+}

diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl_0_9.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl_0_9.java
index 97646a015f..8b1dca77ba 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl_0_9.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl_0_9.java
@@ -36,6 +36,9 @@ public class ServerMethodDispatcherImpl_0_9
 

     private static final BasicRecoverSyncMethodHandler _basicRecoverSyncMethodHandler =

             BasicRecoverSyncMethodHandler.getInstance();

+    private static final QueueUnbindHandler _queueUnbindHandler =

+            QueueUnbindHandler.getInstance();

+

 

     public ServerMethodDispatcherImpl_0_9(AMQStateManager stateManager)

     {

@@ -155,6 +158,7 @@ public class ServerMethodDispatcherImpl_0_9
 

     public boolean dispatchQueueUnbind(QueueUnbindBody body, int channelId) throws AMQException

     {

-        return false;

+        _queueUnbindHandler.methodReceived(getStateManager(),body,channelId);

+        return true;

     }

 }