QPID-3835: update the test preparer tool to generate content for a durable subscription without selector, use it to generate a new legacy store, add test to verify this content once updated and thus expose the issue through test failure

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

3 files changed, 146 insertions, 34 deletions

diff --git a/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBStoreUpgradeTestPreparer.java b/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBStoreUpgradeTestPreparer.java
index 56763fa3bf..bcbb7d8b72 100644
--- a/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBStoreUpgradeTestPreparer.java
+++ b/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBStoreUpgradeTestPreparer.java
@@ -20,8 +20,6 @@
  */
 package org.apache.qpid.server.store.berkeleydb;
 
-import junit.framework.TestCase;
-
 import org.apache.qpid.client.AMQConnectionFactory;
 import org.apache.qpid.url.URLSyntaxException;
 
@@ -29,16 +27,21 @@ import javax.jms.*;
 
 /**
  * Prepares an older version brokers BDB store with the required 
- * contents for use in the BDBStoreUpgradeTest. 
- * 
+ * contents for use in the BDBStoreUpgradeTest.
+ *
+ * NOTE: Must be used with the equivalent older version client!
+ *
  * The store will then be used to verify that the upgraded is 
  * completed properly and that once upgraded it functions as 
  * expected with the new broker.
+ *
  */
-public class BDBStoreUpgradeTestPreparer extends TestCase
+public class BDBStoreUpgradeTestPreparer
 {
     public static final String TOPIC_NAME="myUpgradeTopic";
-    public static final String SUB_NAME="myDurSubName";    
+    public static final String SUB_NAME="myDurSubName";
+    public static final String SELECTOR_SUB_NAME="mySelectorDurSubName";
+    public static final String SELECTOR_TOPIC_NAME="mySelectorUpgradeTopic";
     public static final String QUEUE_NAME="myUpgradeQueue";
 
     private static AMQConnectionFactory _connFac;
@@ -53,19 +56,11 @@ public class BDBStoreUpgradeTestPreparer extends TestCase
         _connFac = new AMQConnectionFactory(CONN_URL);
     }
 
-    /**
-     * Utility test method to allow running the preparation tool
-     * using the test framework
-     */
-    public void testPrepareBroker() throws Exception
-    {
-        prepareBroker();
-    }
-
     private void prepareBroker() throws Exception
     {
         prepareQueues();
-        prepareDurableSubscription();
+        prepareDurableSubscriptionWithSelector();
+        prepareDurableSubscriptionWithoutSelector();
     }
 
     /**
@@ -129,7 +124,7 @@ public class BDBStoreUpgradeTestPreparer extends TestCase
      * - Send a message which matches the selector but will remain uncommitted.
      * - Close the session.
      */
-    private void prepareDurableSubscription() throws Exception
+    private void prepareDurableSubscriptionWithSelector() throws Exception
     {
 
         // Create a connection
@@ -144,10 +139,10 @@ public class BDBStoreUpgradeTestPreparer extends TestCase
         });
         // Create a session on the connection, transacted to confirm delivery
         Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
-        Topic topic = session.createTopic(TOPIC_NAME);
+        Topic topic = session.createTopic(SELECTOR_TOPIC_NAME);
 
         // Create and register a durable subscriber with selector and then close it
-        TopicSubscriber durSub1 = session.createDurableSubscriber(topic, SUB_NAME,"testprop='true'", false);
+        TopicSubscriber durSub1 = session.createDurableSubscriber(topic, SELECTOR_SUB_NAME,"testprop='true'", false);
         durSub1.close();
 
         // Create a publisher and send a persistent message which matches the selector 
@@ -156,14 +151,55 @@ public class BDBStoreUpgradeTestPreparer extends TestCase
         TopicSession pubSession = connection.createTopicSession(true, Session.SESSION_TRANSACTED);
         TopicPublisher publisher = pubSession.createPublisher(topic);
 
-        publishMessages(session, publisher, topic, DeliveryMode.PERSISTENT, 1*1024, 1, "true");
-        publishMessages(session, publisher, topic, DeliveryMode.PERSISTENT, 1*1024, 1, "false");
+        publishMessages(pubSession, publisher, topic, DeliveryMode.PERSISTENT, 1*1024, 1, "true");
+        publishMessages(pubSession, publisher, topic, DeliveryMode.PERSISTENT, 1*1024, 1, "false");
         pubSession.commit();
-        publishMessages(session, publisher, topic, DeliveryMode.PERSISTENT, 1*1024, 1, "true");
+        publishMessages(pubSession, publisher, topic, DeliveryMode.PERSISTENT, 1*1024, 1, "true");
 
         publisher.close();
         pubSession.close();
+        connection.close();
+    }
 
+    /**
+     * Prepare a DurableSubscription backing queue for use in testing use of
+     * DurableSubscriptions without selectors following the upgrade process.
+     *
+     * - Create a transacted session on the connection.
+     * - Open and close a DurableSubscription without selector to create the backing queue.
+     * - Send a message which matches the subscription and commit session.
+     * - Close the session.
+     */
+    private void prepareDurableSubscriptionWithoutSelector() throws Exception
+    {
+        // Create a connection
+        TopicConnection connection = _connFac.createTopicConnection();
+        connection.start();
+        connection.setExceptionListener(new ExceptionListener()
+        {
+            public void onException(JMSException e)
+            {
+                e.printStackTrace();
+            }
+        });
+        // Create a session on the connection, transacted to confirm delivery
+        Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
+        Topic topic = session.createTopic(TOPIC_NAME);
+
+        // Create and register a durable subscriber without selector and then close it
+        TopicSubscriber durSub1 = session.createDurableSubscriber(topic, SUB_NAME);
+        durSub1.close();
+
+        // Create a publisher and send a persistent message which matches the subscription
+        TopicSession pubSession = connection.createTopicSession(true, Session.SESSION_TRANSACTED);
+        TopicPublisher publisher = pubSession.createPublisher(topic);
+
+        publishMessages(pubSession, publisher, topic, DeliveryMode.PERSISTENT, 1*1024, 1, "indifferent");
+        pubSession.commit();
+
+        publisher.close();
+        pubSession.close();
+        connection.close();
     }
 
     public static void sendMessages(Session session, MessageProducer messageProducer, 
diff --git a/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBUpgradeTest.java b/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBUpgradeTest.java
index 227a34e888..55327e3b56 100644
--- a/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBUpgradeTest.java
+++ b/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBUpgradeTest.java
@@ -48,6 +48,8 @@ import org.apache.qpid.util.FileUtils;
 import static org.apache.qpid.server.store.berkeleydb.BDBStoreUpgradeTestPreparer.QUEUE_NAME;
 import static org.apache.qpid.server.store.berkeleydb.BDBStoreUpgradeTestPreparer.SUB_NAME;
 import static org.apache.qpid.server.store.berkeleydb.BDBStoreUpgradeTestPreparer.TOPIC_NAME;
+import static org.apache.qpid.server.store.berkeleydb.BDBStoreUpgradeTestPreparer.SELECTOR_SUB_NAME;
+import static org.apache.qpid.server.store.berkeleydb.BDBStoreUpgradeTestPreparer.SELECTOR_TOPIC_NAME;
 
 import javax.jms.Connection;
 import javax.jms.DeliveryMode;
@@ -154,7 +156,8 @@ public class BDBUpgradeTest extends QpidBrokerTestCase
     /**
      * Test that the selector applied to the DurableSubscription was successfully
      * transfered to the new store, and functions as expected with continued use
-     * by monitoring message count while sending new messages to the topic.
+     * by monitoring message count while sending new messages to the topic and then
+     * consuming them.
      */
     public void testSelectorDurability() throws Exception
     {
@@ -171,7 +174,7 @@ public class BDBUpgradeTest extends QpidBrokerTestCase
 
         try
         {
-            ManagedQueue dursubQueue = jmxUtils.getManagedQueue("clientid" + ":" + SUB_NAME);
+            ManagedQueue dursubQueue = jmxUtils.getManagedQueue("clientid" + ":" + SELECTOR_SUB_NAME);
             assertEquals("DurableSubscription backing queue should have 1 message on it initially", 
                           new Integer(1), dursubQueue.getMessageCount());
             
@@ -181,20 +184,26 @@ public class BDBUpgradeTest extends QpidBrokerTestCase
             
             // Send messages which don't match and do match the selector, checking message count
             TopicSession pubSession = connection.createTopicSession(true, org.apache.qpid.jms.Session.SESSION_TRANSACTED);
-            Topic topic = pubSession.createTopic(TOPIC_NAME);
+            Topic topic = pubSession.createTopic(SELECTOR_TOPIC_NAME);
             TopicPublisher publisher = pubSession.createPublisher(topic);
             
             BDBStoreUpgradeTestPreparer.publishMessages(pubSession, publisher, topic, DeliveryMode.PERSISTENT, 1*1024, 1, "false");
             pubSession.commit();
             assertEquals("DurableSubscription backing queue should still have 1 message on it", 
-                        new Integer(1), dursubQueue.getMessageCount());
+                         Integer.valueOf(1), dursubQueue.getMessageCount());
             
             BDBStoreUpgradeTestPreparer.publishMessages(pubSession, publisher, topic, DeliveryMode.PERSISTENT, 1*1024, 1, "true");
             pubSession.commit();
             assertEquals("DurableSubscription backing queue should now have 2 messages on it", 
-                        new Integer(2), dursubQueue.getMessageCount());
+                         Integer.valueOf(2), dursubQueue.getMessageCount());
+
+            TopicSubscriber durSub = pubSession.createDurableSubscriber(topic, SELECTOR_SUB_NAME,"testprop='true'", false);
+            Message m = durSub.receive(2000);
+            assertNotNull("Failed to receive an expected message", m);
+            m = durSub.receive(2000);
+            assertNotNull("Failed to receive an expected message", m);
+            pubSession.commit();
 
-            dursubQueue.clearQueue();
             pubSession.close();
         }
         finally
@@ -204,6 +213,58 @@ public class BDBUpgradeTest extends QpidBrokerTestCase
     }
 
     /**
+     * Test that the DurableSubscription without selector was successfully
+     * transfered to the new store, and functions as expected with continued use.
+     */
+    public void testDurableSubscriptionWithoutSelector() throws Exception
+    {
+        JMXTestUtils jmxUtils = null;
+        try
+        {
+            jmxUtils = new JMXTestUtils(this, "guest", "guest");
+            jmxUtils.open();
+        }
+        catch (Exception e)
+        {
+            fail("Unable to establish JMX connection, test cannot proceed");
+        }
+
+        try
+        {
+            ManagedQueue dursubQueue = jmxUtils.getManagedQueue("clientid" + ":" + SUB_NAME);
+            assertEquals("DurableSubscription backing queue should have 1 message on it initially",
+                          new Integer(1), dursubQueue.getMessageCount());
+
+            // Create a connection and start it
+            TopicConnection connection = (TopicConnection) getConnection();
+            connection.start();
+
+            // Send new message matching the topic, checking message count
+            TopicSession session = connection.createTopicSession(true, org.apache.qpid.jms.Session.SESSION_TRANSACTED);
+            Topic topic = session.createTopic(TOPIC_NAME);
+            TopicPublisher publisher = session.createPublisher(topic);
+
+            BDBStoreUpgradeTestPreparer.publishMessages(session, publisher, topic, DeliveryMode.PERSISTENT, 1*1024, 1, "indifferent");
+            session.commit();
+            assertEquals("DurableSubscription backing queue should now have 2 messages on it",
+                        Integer.valueOf(2), dursubQueue.getMessageCount());
+
+            TopicSubscriber durSub = session.createDurableSubscriber(topic, SUB_NAME);
+            Message m = durSub.receive(2000);
+            assertNotNull("Failed to receive an expected message", m);
+            m = durSub.receive(2000);
+            assertNotNull("Failed to receive an expected message", m);
+
+            session.commit();
+            session.close();
+        }
+        finally
+        {
+            jmxUtils.close();
+        }
+    }
+
+    /**
      * Test that the backing queue for the durable subscription created was successfully
      * detected and set as being exclusive during the upgrade process, and that the
      * regular queue was not.
@@ -278,7 +339,8 @@ public class BDBUpgradeTest extends QpidBrokerTestCase
         Connection connection = getConnection();
         connection.start();
 
-        consumeDurableSubscriptionMessages(connection);
+        consumeDurableSubscriptionMessages(connection, true);
+        consumeDurableSubscriptionMessages(connection, false);
         consumeQueueMessages(connection, false);
     }
 
@@ -484,21 +546,35 @@ public class BDBUpgradeTest extends QpidBrokerTestCase
         }
     }
 
-    private void consumeDurableSubscriptionMessages(Connection connection) throws Exception
+    private void consumeDurableSubscriptionMessages(Connection connection, boolean selector) throws Exception
     {
         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-        Topic topic = session.createTopic(TOPIC_NAME);
+        Topic topic = null;
+        TopicSubscriber durSub = null;
+
+        if(selector)
+        {
+            topic = session.createTopic(SELECTOR_TOPIC_NAME);
+            durSub = session.createDurableSubscriber(topic, SELECTOR_SUB_NAME,"testprop='true'", false);
+        }
+        else
+        {
+            topic = session.createTopic(TOPIC_NAME);
+            durSub = session.createDurableSubscriber(topic, SUB_NAME);
+        }
 
-        TopicSubscriber durSub = session.createDurableSubscriber(topic, SUB_NAME,"testprop='true'", false);
 
         // Retrieve the matching message 
         Message m = durSub.receive(2000);
         assertNotNull("Failed to receive an expected message", m);
-        assertEquals("Selector property did not match", "true", m.getStringProperty("testprop"));
+        if(selector)
+        {
+            assertEquals("Selector property did not match", "true", m.getStringProperty("testprop"));
+        }
         assertEquals("ID property did not match", 1, m.getIntProperty("ID"));
         assertEquals("Message content was not as expected",BDBStoreUpgradeTestPreparer.generateString(1024) , ((TextMessage)m).getText());
 
-        // Verify that neither the non-matching or uncommitted message are received
+        // Verify that no more messages are received
         m = durSub.receive(1000);
         assertNull("No more messages should have been recieved", m);
 
diff --git a/java/bdbstore/src/test/resources/upgrade/bdbstore-to-upgrade/test-store/00000000.jdb b/java/bdbstore/src/test/resources/upgrade/bdbstore-to-upgrade/test-store/00000000.jdb
index c4e4e6c306..38158a55e7 100644
--- a/java/bdbstore/src/test/resources/upgrade/bdbstore-to-upgrade/test-store/00000000.jdb
+++ b/java/bdbstore/src/test/resources/upgrade/bdbstore-to-upgrade/test-store/00000000.jdb