QPID-6226: [Java Broker] Change queue.declare so that on the passive path, exclusivity is verified if command has set the exclusive flag.

Also strengthen the python tests.

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

2 files changed, 60 insertions, 42 deletions

diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java
index 969701987a..288a4f946c 100644
--- a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java
@@ -53,7 +53,6 @@ import org.apache.qpid.server.model.Queue;
 import org.apache.qpid.server.model.UnknownConfiguredObjectException;
 import org.apache.qpid.server.queue.AMQQueue;
 import org.apache.qpid.server.queue.QueueArgumentsConverter;
-import org.apache.qpid.server.store.DurableConfigurationStore;
 import org.apache.qpid.server.store.MessageStore;
 import org.apache.qpid.server.store.StoreException;
 import org.apache.qpid.server.store.StoreFuture;
@@ -1291,7 +1290,6 @@ public class ServerSessionDelegate extends SessionDelegate
     {
 
         final VirtualHostImpl virtualHost = getVirtualHost(session);
-        DurableConfigurationStore store = virtualHost.getDurableConfigurationStore();
 
         String queueName = method.getQueue();
         AMQQueue queue;
@@ -1312,15 +1310,25 @@ public class ServerSessionDelegate extends SessionDelegate
                 exception(session, method, errorCode, description);
 
             }
-            else if (!verifySessionAccess((ServerSession) session, queue))
+            else if (exclusive)
             {
-                String description = "Cannot declare queue('" + queueName + "'),"
-                                                                       + " as exclusive queue with same name "
-                                                                       + "declared on another session";
-                ExecutionErrorCode errorCode = ExecutionErrorCode.RESOURCE_LOCKED;
-
-                exception(session, method, errorCode, description);
+                if (queue.getExclusive() == ExclusivityPolicy.NONE)
+                {
+                    String description = "Cannot passively declare queue ('" + queueName + "')"
+                                         + " as exclusive as queue with same name is" +
+                                         " already declared as non-exclusive";
+                    ExecutionErrorCode errorCode = ExecutionErrorCode.RESOURCE_LOCKED;
+                    exception(session, method, errorCode, description);
 
+                }
+                else if (!verifySessionAccess((ServerSession) session, queue))
+                {
+                    String description = "Cannot passively declare queue('" + queueName + "'),"
+                                         + " as exclusive queue with same name "
+                                         + "declared on another session";
+                    ExecutionErrorCode errorCode = ExecutionErrorCode.RESOURCE_LOCKED;
+                    exception(session, method, errorCode, description);
+                }
             }
         }
         else
@@ -1387,30 +1395,6 @@ public class ServerSessionDelegate extends SessionDelegate
         }
     }
 
-    /**
-     * Converts a queue argument into a boolean value.  For compatibility with the C++
-     * and the clients, accepts with Boolean, String, or Number types.
-     * @param argValue  argument value.
-     *
-     * @return true if set
-     */
-    private boolean convertBooleanValue(Object argValue)
-    {
-        if(argValue instanceof Boolean && ((Boolean)argValue))
-        {
-            return true;
-        }
-        else if (argValue instanceof String && Boolean.parseBoolean((String)argValue))
-        {
-            return true;
-        }
-        else if (argValue instanceof Number && ((Number)argValue).intValue() != 0)
-        {
-            return true;
-        }
-        return false;
-    }
-
     @Override
     public void queueDelete(Session session, QueueDelete method)
     {
diff --git a/qpid/tests/src/py/qpid_tests/broker_0_10/queue.py b/qpid/tests/src/py/qpid_tests/broker_0_10/queue.py
index 545bc8cc23..132bd7b987 100644
--- a/qpid/tests/src/py/qpid_tests/broker_0_10/queue.py
+++ b/qpid/tests/src/py/qpid_tests/broker_0_10/queue.py
@@ -152,20 +152,54 @@ class QueueTests(TestBase010):
             self.assertEquals(405, e.args[0].error_code)
 
     def test_declare_passive(self):
+      """
+      Test that the passive field is honoured in queue.declare
+      """
+      s1 = self.session
+      s2 = self.conn.session("other")
+
+      s1.queue_declare(queue="passive-queue-1")
+
+      #ensure that same/separate sessions can passively declare same queue
+      s1.queue_declare(queue="passive-queue-1", passive=True)
+      s2.queue_declare(queue="passive-queue-1", passive=True)
+
+      s1.queue_delete(queue="passive-queue-1")
+
+    def test_declare_passive_queue_not_found(self):
+      """
+      Test that the passive field is honoured in queue.declare
+      """
+      s1 = self.session
+
+      try:
+        s1.queue_declare(queue="passive-queue-not-found", passive=True)
+        self.fail("Expected passive declaration of non-existent queue to raise a channel exception")
+      except SessionException, e:
+        self.assertEquals(404, e.args[0].error_code) #not-found
+
+
+    def test_declare_passive_with_exclusive(self):
         """
         Test that the passive field is honoured in queue.declare
         """
-        session = self.session
-        #declare an exclusive queue:
-        session.queue_declare(queue="passive-queue-1", exclusive=True, auto_delete=True)
-        session.queue_declare(queue="passive-queue-1", passive=True)
+        s1 = self.session
+        s2 = self.conn.session("other")
+
+        #declare exclusive/non-exclusive queues:
+        s1.queue_declare(queue="passive-queue-exc", exclusive=True, auto_delete=True)
+        s1.queue_declare(queue="passive-queue-nonexc", exclusive=False, auto_delete=True)
+
+        #ensure that same/separate sessions can passively declare same queue *without* the exclusive flag
+        #this is important for the request/reply pattern
+        s1.queue_declare(queue="passive-queue-exc", passive=True)
+        s2.queue_declare(queue="passive-queue-exc", passive=True)
+
         try:
-            #other connection should not be allowed to declare this:
-            session.queue_declare(queue="passive-queue-2", passive=True)
-            self.fail("Expected passive declaration of non-existant queue to raise a channel exception")
+          s2.queue_declare(queue="passive-queue-nonexc", exclusive=True, passive=True)
+          self.fail("Expected exclusive passive declaration of existing queue to raise a channel exception")
         except SessionException, e:
-            self.assertEquals(404, e.args[0].error_code) #not-found
-
+          self.assertEquals(405, e.args[0].error_code) # resource locked
 
     def test_bind(self):
         """