QPID-5926: [Java Broker] When transitioning from STOPPED to ACTIVE the virtualhost re-recovers children beneath it

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

2 files changed, 59 insertions, 13 deletions

diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/systest/rest/QpidRestTestCase.java b/qpid/java/systests/src/main/java/org/apache/qpid/systest/rest/QpidRestTestCase.java
index 2da4a21b2d..df87432344 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/systest/rest/QpidRestTestCase.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/systest/rest/QpidRestTestCase.java
@@ -112,14 +112,15 @@ public class QpidRestTestCase extends QpidBrokerTestCase
     public Map<String, Object> waitForAttributeChanged(String url, String attributeName, Object newValue) throws Exception
     {
         List<Map<String, Object>> nodeAttributes = getRestTestHelper().getJsonAsList(url);
-        long limit = System.currentTimeMillis() + 5000;
+        int timeout = 5000;
+        long limit = System.currentTimeMillis() + timeout;
         while(System.currentTimeMillis() < limit && (nodeAttributes.size() == 0 || !newValue.equals(nodeAttributes.get(0).get(attributeName))))
         {
             Thread.sleep(100l);
             nodeAttributes = getRestTestHelper().getJsonAsList(url);
         }
         Map<String, Object> nodeData = nodeAttributes.get(0);
-        assertEquals("Unexpected attribute " + attributeName, newValue, nodeData.get(attributeName));
+        assertEquals("Attribute " + attributeName + " did not reach expected value within permitted timeout "  + timeout + "ms.", newValue, nodeData.get(attributeName));
         return nodeData;
     }
 }
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/systest/rest/VirtualHostRestTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/systest/rest/VirtualHostRestTest.java
index da89772a22..243b93e798 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/systest/rest/VirtualHostRestTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/systest/rest/VirtualHostRestTest.java
@@ -27,12 +27,11 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import javax.jms.Connection;
+import javax.jms.Destination;
 import javax.jms.Session;
 import javax.servlet.http.HttpServletResponse;
 
-import org.codehaus.jackson.JsonGenerationException;
-import org.codehaus.jackson.map.JsonMappingException;
-
 import org.apache.qpid.server.virtualhost.ProvidedStoreVirtualHostImpl;
 import org.apache.qpid.server.virtualhostnode.JsonVirtualHostNode;
 import org.apache.qpid.client.AMQConnection;
@@ -163,22 +162,60 @@ public class VirtualHostRestTest extends QpidRestTestCase
 
     public void testMutateState() throws Exception
     {
-        String hostToUpdate = TEST3_VIRTUALHOST;
-        String restHostUrl = "virtualhost/" + hostToUpdate + "/" + hostToUpdate;
+        String restHostUrl = "virtualhost/" + TEST1_VIRTUALHOST + "/" + TEST1_VIRTUALHOST;
 
+        waitForAttributeChanged(restHostUrl, VirtualHost.STATE, "ACTIVE");
         assertActualAndDesireStates(restHostUrl, "ACTIVE", "ACTIVE");
 
         Map<String, Object> newAttributes = Collections.<String, Object>singletonMap(VirtualHost.DESIRED_STATE, "STOPPED");
         getRestTestHelper().submitRequest(restHostUrl, "PUT", newAttributes, HttpServletResponse.SC_OK);
 
+        waitForAttributeChanged(restHostUrl, VirtualHost.STATE, "STOPPED");
         assertActualAndDesireStates(restHostUrl, "STOPPED", "STOPPED");
 
         newAttributes = Collections.<String, Object>singletonMap(VirtualHost.DESIRED_STATE, "ACTIVE");
         getRestTestHelper().submitRequest(restHostUrl, "PUT", newAttributes, HttpServletResponse.SC_OK);
 
+        waitForAttributeChanged(restHostUrl, VirtualHost.STATE, "ACTIVE");
+
         assertActualAndDesireStates(restHostUrl, "ACTIVE", "ACTIVE");
     }
 
+    public void testMutateStateOfVirtualHostWithQueuesAndMessages() throws Exception
+    {
+        String testQueueName = getTestQueueName();
+        String restHostUrl = "virtualhost/" + TEST1_VIRTUALHOST + "/" + TEST1_VIRTUALHOST;
+        String restQueueUrl = "queue/" + TEST1_VIRTUALHOST + "/" + TEST1_VIRTUALHOST + "/" + testQueueName;
+
+        waitForAttributeChanged(restHostUrl, VirtualHost.STATE, "ACTIVE");
+        assertActualAndDesireStates(restHostUrl, "ACTIVE", "ACTIVE");
+
+        Connection connection = getConnection();
+        Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
+        Destination dest = session.createQueue(testQueueName);
+        session.createConsumer(dest).close();
+        session.createProducer(dest).send(session.createTextMessage("My test message"));
+        session.commit();
+        connection.close();
+
+        assertQueueDepth(restQueueUrl, "Unexpected number of messages before stopped", 1);
+
+        Map<String, Object> newAttributes = Collections.<String, Object>singletonMap(VirtualHost.DESIRED_STATE, "STOPPED");
+        getRestTestHelper().submitRequest(restHostUrl, "PUT", newAttributes, HttpServletResponse.SC_OK);
+
+        waitForAttributeChanged(restHostUrl, VirtualHost.STATE, "STOPPED");
+        assertActualAndDesireStates(restHostUrl, "STOPPED", "STOPPED");
+
+        newAttributes = Collections.<String, Object>singletonMap(VirtualHost.DESIRED_STATE, "ACTIVE");
+        getRestTestHelper().submitRequest(restHostUrl, "PUT", newAttributes, HttpServletResponse.SC_OK);
+
+        waitForAttributeChanged(restHostUrl, VirtualHost.STATE, "ACTIVE");
+
+        assertActualAndDesireStates(restHostUrl, "ACTIVE", "ACTIVE");
+
+        assertQueueDepth(restQueueUrl, "Unexpected number of messages after restart", 1);
+    }
+
     public void testRecoverVirtualHostInDesiredStateStoppedWithDescription() throws Exception
     {
         String hostToUpdate = TEST3_VIRTUALHOST;
@@ -502,15 +539,13 @@ public class VirtualHostRestTest extends QpidRestTestCase
         assertEquals("Unexpected response code", 201, statusCode);
     }
 
-    private void createQueue(String queueName, String queueType, Map<String, Object> attributes) throws IOException,
-            JsonGenerationException, JsonMappingException
+    private void createQueue(String queueName, String queueType, Map<String, Object> attributes) throws Exception
     {
         int responseCode = tryCreateQueue(queueName, queueType, attributes);
         assertEquals("Unexpected response code", 201, responseCode);
     }
 
-    private int tryCreateQueue(String queueName, String queueType, Map<String, Object> attributes) throws IOException,
-            JsonGenerationException, JsonMappingException
+    private int tryCreateQueue(String queueName, String queueType, Map<String, Object> attributes) throws Exception
     {
         Map<String, Object> queueData = new HashMap<String, Object>();
         queueData.put(Queue.NAME, queueName);
@@ -580,11 +615,21 @@ public class VirtualHostRestTest extends QpidRestTestCase
     }
 
     private void assertActualAndDesireStates(final String restUrl,
-                                                            final String expectedDesiredState,
-                                                            final String expectedActualState) throws IOException
+                                             final String expectedDesiredState,
+                                             final String expectedActualState) throws IOException
     {
         Map<String, Object> virtualhost = getRestTestHelper().getJsonAsSingletonList(restUrl);
         Asserts.assertActualAndDesiredState(expectedDesiredState, expectedActualState, virtualhost);
     }
 
+    private void assertQueueDepth(String restQueueUrl, String message, int expectedDepth) throws IOException
+    {
+        Map<String, Object> queueDetails = getRestTestHelper().getJsonAsSingletonList(restQueueUrl);
+        assertNotNull(queueDetails);
+        Map<String, Object> statistics = (Map<String, Object>) queueDetails.get(Asserts.STATISTICS_ATTRIBUTE);
+        assertNotNull(statistics);
+
+        assertEquals(message, expectedDepth, statistics.get("queueDepthMessages"));
+    }
+
 }