QPID-2442: Make 0-10 TemporaryQueue#delete and TemporaryTopic#delete perform the delete on the Broker

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

3 files changed, 276 insertions, 283 deletions

diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/temporaryqueue/TemporaryQueueTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/temporaryqueue/TemporaryQueueTest.java
index 8c806fa2da..c98e403671 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/temporaryqueue/TemporaryQueueTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/temporaryqueue/TemporaryQueueTest.java
@@ -22,237 +22,145 @@
 package org.apache.qpid.test.unit.client.temporaryqueue;
 
 import javax.jms.Connection;
-import javax.jms.ExceptionListener;
 import javax.jms.JMSException;
 import javax.jms.MessageConsumer;
 import javax.jms.MessageProducer;
 import javax.jms.Session;
 import javax.jms.TemporaryQueue;
 import javax.jms.TextMessage;
-import junit.framework.Assert;
 
+import org.apache.qpid.client.AMQDestination;
+import org.apache.qpid.client.AMQSession;
 import org.apache.qpid.test.utils.QpidBrokerTestCase;
-import org.apache.qpid.client.AMQQueue;
-import org.apache.qpid.jms.ConnectionListener;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.LinkedList;
-
-public class TemporaryQueueTest extends QpidBrokerTestCase implements ExceptionListener
+/**
+ * Tests the behaviour of {@link TemporaryQueue}.
+ */
+public class TemporaryQueueTest extends QpidBrokerTestCase
 {
-    private List<Exception> _exceptions = new ArrayList<Exception>();
-
-    protected void setUp() throws Exception
-    {
-        super.setUp();
-    }
-
-    protected void tearDown() throws Exception
+    /**
+     * Tests the basic produce/consume behaviour of a temporary queue.
+     */
+    public void testMessageDeliveryUsingTemporaryQueue() throws Exception
     {
-       super.tearDown();
-    }
-
-    protected Connection createConnection() throws Exception
-    {
-        return  getConnection("guest", "guest");
-    }
-
-    public void testTemporaryQueue() throws Exception
-    {
-        Connection conn = createConnection();
-        Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
-        TemporaryQueue queue = session.createTemporaryQueue();
+        final Connection conn = getConnection();
+        final Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        final TemporaryQueue queue = session.createTemporaryQueue();
         assertNotNull(queue);
-        MessageProducer producer = session.createProducer(queue);
-        MessageConsumer consumer = session.createConsumer(queue);
+        final MessageProducer producer = session.createProducer(queue);
+        final MessageConsumer consumer = session.createConsumer(queue);
         conn.start();
         producer.send(session.createTextMessage("hello"));
         TextMessage tm = (TextMessage) consumer.receive(2000);
-        assertNotNull(tm);
+        assertNotNull("Message not received", tm);
         assertEquals("hello", tm.getText());
+    }
 
-        try
-        {
-            queue.delete();
-            fail("Expected JMSException : should not be able to delete while there are active consumers");
-        }
-        catch (JMSException je)
-        {
-            ; //pass
-        }
-
-        consumer.close();
+    /**
+     * Tests that a temporary queue cannot be used by another {@link Session}.
+     */
+    public void testUseFromAnotherSessionProhibited() throws Exception
+    {
+        final Connection conn = getConnection();
+        final Session session1 = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        final Session session2 = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        final TemporaryQueue queue = session1.createTemporaryQueue();
+        assertNotNull(queue);
 
         try
         {
-            queue.delete();
+            session2.createConsumer(queue);
+            fail("Expected a JMSException when subscribing to a temporary queue created on a different session");
         }
         catch (JMSException je)
         {
-            fail("Unexpected Exception: " + je.getMessage());
+            //pass
+            assertEquals("Cannot consume from a temporary destination created on another session", je.getMessage());
         }
-
-        conn.close();
-    }
-
-    public void tUniqueness() throws Exception
-    {
-        int numProcs = Runtime.getRuntime().availableProcessors();
-        final int threadsProc = 5;
-
-        runUniqueness(1, 10);
-        runUniqueness(numProcs * threadsProc, 10);
-        runUniqueness(numProcs * threadsProc, 100);
-        runUniqueness(numProcs * threadsProc, 500);
     }
 
-    void runUniqueness(int makers, int queues) throws Exception
+    /**
+     * Tests that the client is able to explicitly delete a temporary queue using
+     * {@link TemporaryQueue#delete()} and is prevented from deleting one that
+     * still has consumers.
+     *
+     * Note: Under < 0-10 {@link TemporaryQueue#delete()} only marks the queue as deleted
+     * on the client. 0-10 causes the queue to be deleted from the Broker.
+     */
+    public void testExplictTemporaryQueueDeletion() throws Exception
     {
-        Connection connection = createConnection();
-
-        Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
-
-        List<TempQueueMaker> tqList = new LinkedList<TempQueueMaker>();
-
-        //Create Makers
-        for (int m = 0; m < makers; m++)
-        {
-            tqList.add(new TempQueueMaker(session, queues));
-        }
-
-
-        List<Thread> threadList = new LinkedList<Thread>();
-
-        //Create Makers
-        for (TempQueueMaker maker : tqList)
-        {
-            threadList.add(new Thread(maker));
-        }
-
-        //Start threads
-        for (Thread thread : threadList)
-        {
-            thread.start();
-        }
-
-        // Join Threads
-        for (Thread thread : threadList)
-        {
-            try
-            {
-                thread.join();
-            }
-            catch (InterruptedException e)
-            {
-                fail("Couldn't correctly join threads");
-            }
-        }
-
+        final Connection conn = getConnection();
+        final Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        final AMQSession<?, ?> amqSession = (AMQSession<?, ?>)session; // Required to observe the queue binding on the Broker
+        final TemporaryQueue queue = session.createTemporaryQueue();
+        assertNotNull(queue);
+        final MessageConsumer consumer = session.createConsumer(queue);
+        conn.start();
 
-        List<AMQQueue> list = new LinkedList<AMQQueue>();
+        assertTrue("Queue should be bound", amqSession.isQueueBound((AMQDestination)queue));
 
-        // Test values
-        for (TempQueueMaker maker : tqList)
+        try
         {
-            check(maker, list);
+            queue.delete();
+            fail("Expected JMSException : should not be able to delete while there are active consumers");
         }
-
-        Assert.assertEquals("Not enough queues made.", makers * queues, list.size());
-
-        connection.close();
-    }
-
-    private void check(TempQueueMaker tq, List<AMQQueue> list)
-    {
-        for (AMQQueue q : tq.getList())
+        catch (JMSException je)
         {
-            if (list.contains(q))
-            {
-                fail(q + " already exists.");
-            }
-            else
-            {
-                list.add(q);
-            }
+            //pass
+            assertEquals("Temporary Queue has consumers so cannot be deleted", je.getMessage());
         }
-    }
-
-
-    class TempQueueMaker implements Runnable
-    {
-        List<AMQQueue> _queues;
-        Session _session;
-        private int _count;
+        consumer.close();
 
+        // Now deletion should succeed.
+        queue.delete();
 
-        TempQueueMaker(Session session, int queues) throws JMSException
+        try
         {
-            _queues = new LinkedList<AMQQueue>();
-
-            _count = queues;
-
-            _session = session;
+            session.createConsumer(queue);
+            fail("Exception not thrown");
         }
-
-        public void run()
+        catch (JMSException je)
         {
-            int i = 0;
-            try
-            {
-                for (; i < _count; i++)
-                {
-                    _queues.add((AMQQueue) _session.createTemporaryQueue());
-                }
-            }
-            catch (JMSException jmse)
-            {
-                //stop
-            }
+            //pass
+            assertEquals("Cannot consume from a deleted destination", je.getMessage());
         }
 
-        List<AMQQueue> getList()
+        if (isBroker010())
         {
-            return _queues;
+            assertFalse("Queue should no longer be bound", amqSession.isQueueBound((AMQDestination)queue));
         }
     }
 
-    public void testQPID1217() throws Exception
-    {
-        Connection conA = getConnection();
-        conA.setExceptionListener(this);
-        Session sessA = conA.createSession(false, Session.AUTO_ACKNOWLEDGE);
-        TemporaryQueue temp = sessA.createTemporaryQueue();
-        
-        MessageProducer prod = sessA.createProducer(temp);
-        prod.send(sessA.createTextMessage("hi"));
-
-        Thread.sleep(500);
-        assertTrue("Exception received", _exceptions.isEmpty());
-        
-        Connection conB = getConnection();
-        Session sessB = conB.createSession(false, Session.AUTO_ACKNOWLEDGE);
-        
-        JMSException ex = null;
-        try
-        {
-            MessageConsumer consB = sessB.createConsumer(temp);
-        } 
-        catch (JMSException e)
-        {
-            ex = e; 
-        }
-        assertNotNull(ex);
-    }
-    
-    public static junit.framework.Test suite()
+    /**
+     * Tests that a temporary queue remains available for reuse even after its initial
+     * consumer has disconnected.
+     *
+     *  This test would fail under < 0-10 as their temporary queues are deleted automatically
+     *  (broker side) after the last consumer disconnects (so message2 would be lost). For this
+     *  reason this test is excluded from those profiles.
+     */
+    public void testTemporaryQueueReused() throws Exception
     {
-        return new junit.framework.TestSuite(TemporaryQueueTest.class);
-    }
+        final Connection conn = getConnection();
+        final Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        final TemporaryQueue queue = session.createTemporaryQueue();
+        assertNotNull(queue);
 
-    public void onException(JMSException arg0)
-    {
-        _exceptions.add(arg0);
+        final MessageProducer producer1 = session.createProducer(queue);
+        final MessageConsumer consumer1 = session.createConsumer(queue);
+        conn.start();
+        producer1.send(session.createTextMessage("message1"));
+        producer1.send(session.createTextMessage("message2"));
+        TextMessage tm = (TextMessage) consumer1.receive(2000);
+        assertNotNull("Message not received by first consumer", tm);
+        assertEquals("message1", tm.getText());
+        consumer1.close();
+
+        final MessageConsumer consumer2 = session.createConsumer(queue);
+        conn.start();
+        tm = (TextMessage) consumer2.receive(2000);
+        assertNotNull("Message not received by second consumer", tm);
+        assertEquals("message2", tm.getText());
+        consumer2.close();
     }
-
 }
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/topic/TemporaryTopicTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/topic/TemporaryTopicTest.java
new file mode 100644
index 0000000000..c89b13a0f9
--- /dev/null
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/topic/TemporaryTopicTest.java
@@ -0,0 +1,182 @@
+/*
+ *
+ * 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.test.unit.topic;
+
+import javax.jms.Connection;
+import javax.jms.JMSException;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.TemporaryTopic;
+import javax.jms.TextMessage;
+
+import org.apache.qpid.test.utils.QpidBrokerTestCase;
+
+
+/**
+ * Tests the behaviour of {@link TemporaryTopic}.
+ */
+public class TemporaryTopicTest extends QpidBrokerTestCase
+{
+    /**
+     * Tests the basic publish/subscribe behaviour of a temporary topic. Single
+     * message is sent to two subscribers.
+     */
+    public void testMessageDeliveryUsingTemporaryTopic() throws Exception
+    {
+        final Connection conn = getConnection();
+        final Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        final TemporaryTopic topic = session.createTemporaryTopic();
+        assertNotNull(topic);
+        final MessageProducer producer = session.createProducer(topic);
+        final MessageConsumer consumer1 = session.createConsumer(topic);
+        final MessageConsumer consumer2 = session.createConsumer(topic);
+        conn.start();
+        producer.send(session.createTextMessage("hello"));
+
+        final TextMessage tm1 = (TextMessage) consumer1.receive(2000);
+        final TextMessage tm2 = (TextMessage) consumer2.receive(2000);
+
+        assertNotNull("Message not received by subscriber1", tm1);
+        assertEquals("hello", tm1.getText());
+        assertNotNull("Message not received by subscriber2", tm2);
+        assertEquals("hello", tm2.getText());
+    }
+
+    /**
+     * Tests that the client is able to explicitly delete a temporary topic using
+     * {@link TemporaryTopic#delete()} and is prevented from deleting one that
+     * still has consumers.
+     *
+     * Note: Under < 0-10 {@link TemporaryTopic#delete()} only marks the queue as deleted
+     * on the client. 0-10 causes the topic to be deleted from the Broker.
+     */
+    public void testExplictTemporaryTopicDeletion() throws Exception
+    {
+        final Connection conn = getConnection();
+
+        final Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        final TemporaryTopic topic = session.createTemporaryTopic();
+        assertNotNull(topic);
+        final MessageConsumer consumer = session.createConsumer(topic);
+        conn.start();
+        try
+        {
+            topic.delete();
+            fail("Expected JMSException : should not be able to delete while there are active consumers");
+        }
+        catch (JMSException je)
+        {
+            //pass
+            assertEquals("Temporary Topic has consumers so cannot be deleted", je.getMessage());
+        }
+
+        consumer.close();
+
+        // Now deletion should succeed.
+        topic.delete();
+
+        try
+        {
+            session.createConsumer(topic);
+            fail("Exception not thrown");
+        }
+        catch (JMSException je)
+        {
+            //pass
+            assertEquals("Cannot consume from a deleted destination", je.getMessage());
+        }
+    }
+
+    /**
+     * Tests that a temporary topic cannot be used by another {@link Session}.
+     */
+    public void testUseFromAnotherSessionProhibited() throws Exception
+    {
+        final Connection conn = getConnection();
+        final Session session1 = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        final Session session2 = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        final TemporaryTopic topic = session1.createTemporaryTopic();
+
+        try
+        {
+            session2.createConsumer(topic);
+            fail("Expected a JMSException when subscribing to a temporary topic created on a different session");
+        }
+        catch (JMSException je)
+        {
+            // pass
+            assertEquals("Cannot consume from a temporary destination created on another session", je.getMessage());
+        }
+    }
+
+    /**
+     * Tests that the client is prohibited from creating a durable subscriber for a temporary
+     * queue.
+     */
+    public void testDurableSubscriptionProhibited() throws Exception
+    {
+        final Connection conn = getConnection();
+
+        final Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        final TemporaryTopic topic = session.createTemporaryTopic();
+        assertNotNull(topic);
+        try
+        {
+            session.createDurableSubscriber(topic, null);
+            fail("Expected JMSException : should not be able to create durable subscription from temp topic");
+        }
+        catch (JMSException je)
+        {
+            //pass
+            assertEquals("Cannot create a durable subscription with a temporary topic: " + topic.toString(), je.getMessage());
+        }
+    }
+
+    /**
+     * Tests that a temporary topic remains available for reuse even after its initial
+     * subscribers have disconnected.
+     */
+    public void testTemporaryTopicReused() throws Exception
+    {
+        final Connection conn = getConnection();
+        final Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        final TemporaryTopic topic = session.createTemporaryTopic();
+        assertNotNull(topic);
+
+        final MessageProducer producer = session.createProducer(topic);
+        final MessageConsumer consumer1 = session.createConsumer(topic);
+        conn.start();
+        producer.send(session.createTextMessage("message1"));
+        TextMessage tm = (TextMessage) consumer1.receive(2000);
+        assertNotNull("Message not received by first consumer", tm);
+        assertEquals("message1", tm.getText());
+        consumer1.close();
+
+        final MessageConsumer consumer2 = session.createConsumer(topic);
+        conn.start();
+        producer.send(session.createTextMessage("message2"));
+        tm = (TextMessage) consumer2.receive(2000);
+        assertNotNull("Message not received by second consumer", tm);
+        assertEquals("message2", tm.getText());
+        consumer2.close();
+    }
+}
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java
index eee232e113..0b1aeef8e9 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java
@@ -21,10 +21,7 @@
 package org.apache.qpid.test.unit.topic;
 
 import javax.jms.InvalidDestinationException;
-import javax.jms.JMSException;
 import javax.jms.MessageConsumer;
-import javax.jms.Session;
-import javax.jms.TemporaryTopic;
 import javax.jms.TextMessage;
 import javax.jms.TopicPublisher;
 import javax.jms.TopicSession;
@@ -40,18 +37,6 @@ import org.apache.qpid.test.utils.QpidBrokerTestCase;
 /** @author Apache Software Foundation */
 public class TopicSessionTest extends QpidBrokerTestCase
 {
-
-    protected void setUp() throws Exception
-    {
-        super.setUp();
-    }
-
-    protected void tearDown() throws Exception
-    {
-        super.tearDown();
-    }
-
-
     public void testTopicSubscriptionUnsubscription() throws Exception
     {
 
@@ -228,83 +213,6 @@ public class TopicSessionTest extends QpidBrokerTestCase
         con.close();
     }
 
-    public void testSendingSameMessage() throws Exception
-    {
-        AMQConnection conn = (AMQConnection) getConnection("guest", "guest");
-        TopicSession session = conn.createTopicSession(true, Session.AUTO_ACKNOWLEDGE);
-        TemporaryTopic topic = session.createTemporaryTopic();
-        assertNotNull(topic);
-        TopicPublisher producer = session.createPublisher(topic);
-        MessageConsumer consumer = session.createConsumer(topic);
-        conn.start();
-        TextMessage sentMessage = session.createTextMessage("Test Message");
-        producer.send(sentMessage);
-        session.commit();
-        TextMessage receivedMessage = (TextMessage) consumer.receive(2000);
-        assertNotNull(receivedMessage);
-        assertEquals(sentMessage.getText(), receivedMessage.getText());
-        producer.send(sentMessage);
-        session.commit();
-        receivedMessage = (TextMessage) consumer.receive(2000);
-        assertNotNull(receivedMessage);
-        assertEquals(sentMessage.getText(), receivedMessage.getText());
-        session.commit();
-        conn.close();
-
-    }
-
-    public void testTemporaryTopic() throws Exception
-    {
-        AMQConnection conn = (AMQConnection) getConnection("guest", "guest");
-        TopicSession session = conn.createTopicSession(true, Session.AUTO_ACKNOWLEDGE);
-        TemporaryTopic topic = session.createTemporaryTopic();
-        assertNotNull(topic);
-        TopicPublisher producer = session.createPublisher(topic);
-        MessageConsumer consumer = session.createConsumer(topic);
-        conn.start();
-        producer.send(session.createTextMessage("hello"));
-        session.commit();
-        TextMessage tm = (TextMessage) consumer.receive(2000);
-        assertNotNull(tm);
-        assertEquals("hello", tm.getText());
-        session.commit();
-        try
-        {
-            topic.delete();
-            fail("Expected JMSException : should not be able to delete while there are active consumers");
-        }
-        catch (JMSException je)
-        {
-            ; //pass
-        }
-
-        consumer.close();
-
-        try
-        {
-            topic.delete();
-        }
-        catch (JMSException je)
-        {
-            fail("Unexpected Exception: " + je.getMessage());
-        }
-
-        TopicSession session2 = conn.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
-        try
-        {
-            session2.createConsumer(topic);
-            fail("Expected a JMSException when subscribing to a temporary topic created on adifferent session");
-        }
-        catch (JMSException je)
-        {
-            ; // pass
-        }
-
-
-        conn.close();
-    }
-
-
     public void testNoLocal() throws Exception
     {
 
@@ -445,9 +353,4 @@ public class TopicSessionTest extends QpidBrokerTestCase
         assertEquals("Queue depth was wrong", 0, depth);
 
     }
-
-    public static junit.framework.Test suite()
-    {
-        return new junit.framework.TestSuite(TopicSessionTest.class);
-    }
 }