QPID-5944: [Java Broker] Expose queue clear management operation to the REST API and Web Management Console

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

8 files changed, 132 insertions, 45 deletions

diff --git a/qpid/doc/book/src/java-broker/Java-Broker-Security-ACLs.xml b/qpid/doc/book/src/java-broker/Java-Broker-Security-ACLs.xml
index bd0d543c05..1ad7f6715c 100644
--- a/qpid/doc/book/src/java-broker/Java-Broker-Security-ACLs.xml
+++ b/qpid/doc/book/src/java-broker/Java-Broker-Security-ACLs.xml
@@ -617,7 +617,7 @@ ACL ALLOW-LOG webadmins CREATE USER
 ACL ALLOW-LOG webadmins DELETE USER
 ACL ALLOW-LOG webadmins UPDATE USER
 
-# allow to the users from webadmins group to move, copy and delete messagaes
+# allow to the users from webadmins group to move, copy, delete messagaes, and clear the queue
 # using REST management interfaces
 ACL ALLOW-LOG webadmins UPDATE METHOD
 
@@ -625,6 +625,7 @@ ACL ALLOW-LOG webadmins UPDATE METHOD
 #ACL ALLOW-LOG webadmins UPDATE METHOD component="VirtualHost.Queue" name="moveMessages"
 #ACL ALLOW-LOG webadmins UPDATE METHOD component="VirtualHost.Queue" name="copyMessages"
 #ACL ALLOW-LOG webadmins UPDATE METHOD component="VirtualHost.Queue" name="deleteMessages"
+#ACL ALLOW-LOG webadmins UPDATE METHOD component="VirtualHost.Queue" name="clearQueue"
 
 ACL DENY-LOG all all
       </programlisting>
diff --git a/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/servlet/rest/MessageServlet.java b/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/servlet/rest/MessageServlet.java
index 3ee53ad892..531ea1e3c1 100644
--- a/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/servlet/rest/MessageServlet.java
+++ b/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/servlet/rest/MessageServlet.java
@@ -170,7 +170,7 @@ public class MessageServlet extends AbstractServlet
             _messageIds = messageIds;
         }
 
-
+        @Override
         public void withinTransaction(final VirtualHost.Transaction txn)
         {
 
@@ -208,6 +208,7 @@ public class MessageServlet extends AbstractServlet
             _destinationQueue = destinationQueue;
         }
 
+        @Override
         protected void updateEntry(QueueEntry entry, VirtualHost.Transaction txn)
         {
             txn.move(entry, _destinationQueue);
@@ -224,6 +225,7 @@ public class MessageServlet extends AbstractServlet
             _destinationQueue = destinationQueue;
         }
 
+        @Override
         protected void updateEntry(QueueEntry entry, VirtualHost.Transaction txn)
         {
             txn.copy(entry, _destinationQueue);
@@ -237,6 +239,7 @@ public class MessageServlet extends AbstractServlet
             super(sourceQueue, messageIds);
         }
 
+        @Override
         protected void updateEntry(QueueEntry entry, VirtualHost.Transaction txn)
         {
             txn.dequeue(entry);
@@ -244,6 +247,34 @@ public class MessageServlet extends AbstractServlet
     }
 
 
+    private static class ClearQueueTransaction implements VirtualHost.TransactionalOperation
+    {
+        private final Queue _queue;
+
+        protected ClearQueueTransaction(Queue queue)
+        {
+            _queue = queue;
+        }
+
+        @Override
+        public void withinTransaction(final VirtualHost.Transaction txn)
+        {
+            _queue.visit(new QueueEntryVisitor()
+            {
+
+                public boolean visit(final QueueEntry entry)
+                {
+                    final ServerMessage message = entry.getMessage();
+                    if(message != null)
+                    {
+                        txn.dequeue(entry);
+                    }
+                    return false;
+                }
+            });
+
+        }
+    }
 
     private class MessageCollector implements QueueEntryVisitor
     {
@@ -426,7 +457,7 @@ public class MessageServlet extends AbstractServlet
 
 
             final Queue destinationQueue = getQueueFromVirtualHost(destQueueName, vhost);
-            final List messageIds = new ArrayList((List) providedObject.get("messages"));
+            final List<Long> messageIds = new ArrayList<Long>((List<Long>) providedObject.get("messages"));
             QueueEntryTransaction txn =
                     isMoveTransaction
                             ? new MoveTransaction(sourceQueue, messageIds, destinationQueue)
@@ -446,31 +477,33 @@ public class MessageServlet extends AbstractServlet
         }
     }
 
-
     /*
-     * DELETE removes messages from the queue
+     * DELETE removes specified messages from, or clears the queue
      */
     @Override
-    protected void doDeleteWithSubjectAndActor(HttpServletRequest request, HttpServletResponse response)
+    protected void doDeleteWithSubjectAndActor(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException
     {
+        final Queue<?> queue = getQueueFromRequest(request);
 
-        final Queue<?> sourceQueue = getQueueFromRequest(request);
-
-        final VirtualHost<?,?,?> vhost = sourceQueue.getParent(VirtualHost.class);
+        final VirtualHost<?,?,?> vhost = queue.getParent(VirtualHost.class);
+        boolean clearQueue = Boolean.parseBoolean(request.getParameter("clear"));
 
-
-        final List<Long> messageIds = new ArrayList<Long>();
-        for(String idStr : request.getParameterValues("id"))
-        {
-            messageIds.add(Long.valueOf(idStr));
-        }
-
-        // FIXME: added temporary authorization check until we introduce management layer
-        // and review current ACL rules to have common rules for all management interfaces
         try
         {
-            authorizeMethod("deleteMessages", vhost);
-            vhost.executeTransaction(new DeleteTransaction(sourceQueue, messageIds));
+            if (clearQueue)
+            {
+                clearQueue(queue, vhost);
+            }
+            else
+            {
+                final List<Long> messageIds = new ArrayList<>();
+                for(String idStr : request.getParameterValues("id"))
+                {
+                    messageIds.add(Long.valueOf(idStr));
+                }
+
+                deleteMessages(queue, vhost, messageIds);
+            }
             response.setStatus(HttpServletResponse.SC_OK);
         }
         catch (AccessControlException e)
@@ -480,6 +513,22 @@ public class MessageServlet extends AbstractServlet
 
     }
 
+    private void deleteMessages(final Queue<?> queue, final VirtualHost<?, ?, ?> vhost, final List<Long> messageIds)
+    {
+        // FIXME: added temporary authorization check until we introduce management layer
+        // and review current ACL rules to have common rules for all management interfaces
+        authorizeMethod("deleteMessages", vhost);
+        vhost.executeTransaction(new DeleteTransaction(queue, messageIds));
+    }
+
+    private void clearQueue(final Queue<?> queue, final VirtualHost<?, ?, ?> vhost)
+    {
+        // FIXME: added temporary authorization check until we introduce management layer
+        // and review current ACL rules to have common rules for all management interfaces
+        authorizeMethod("clearQueue", vhost);
+        vhost.executeTransaction(new ClearQueueTransaction(queue));
+    }
+
     private void authorizeMethod(String methodName, VirtualHost<?,?,?> vhost)
     {
         SecurityManager securityManager = getBroker().getSecurityManager();
diff --git a/qpid/java/broker-plugins/management-http/src/main/java/resources/js/qpid/management/Queue.js b/qpid/java/broker-plugins/management-http/src/main/java/resources/js/qpid/management/Queue.js
index 3371445e5b..25c0740730 100644
--- a/qpid/java/broker-plugins/management-http/src/main/java/resources/js/qpid/management/Queue.js
+++ b/qpid/java/broker-plugins/management-http/src/main/java/resources/js/qpid/management/Queue.js
@@ -135,6 +135,13 @@ define(["dojo/_base/xhr",
                                                 event.stop(evt);
                                                 that.deleteMessages();
                                             });
+                            var clearQueueButton = query(".clearQueueButton", contentPane.containerNode)[0];
+                            var clearQueueWidget = registry.byNode(clearQueueButton);
+                            connect.connect(clearQueueWidget, "onClick",
+                                function(evt){
+                                    event.stop(evt);
+                                    that.clearQueue();
+                                });
                             var moveMessagesButton = query(".moveMessagesButton", contentPane.containerNode)[0];
                             connect.connect(registry.byNode(moveMessagesButton), "onClick",
                                             function(evt){
@@ -203,7 +210,24 @@ define(["dojo/_base/xhr",
                    }
                }
            };
-
+           Queue.prototype.clearQueue = function() {
+               var that = this;
+               if(confirm("Clear all messages from queue?")) {
+                   var query = "service/message/"+ encodeURIComponent(that.getVirtualHostName())
+                       + "/" + encodeURIComponent(that.getQueueName()) + "?clear=true";
+                   that.success = true
+                   xhr.del({url: query, sync: true, handleAs: "json"}).then(
+                       function(data) {
+                           that.grid.setQuery({id: "*"});
+                           that.grid.selection.deselectAll();
+                           that.queueUpdater.update();
+                       },
+                       function(error) {that.success = false; that.failureReason = error;});
+                   if(!that.success ) {
+                       alert("Error:" + this.failureReason);
+                   }
+               }
+           };
            Queue.prototype.moveOrCopyMessages = function(obj) {
                var that = this;
                var move = obj.move;
diff --git a/qpid/java/broker-plugins/management-http/src/main/java/resources/js/qpid/management/authenticationprovider/PrincipalDatabaseAuthenticationManager.js b/qpid/java/broker-plugins/management-http/src/main/java/resources/js/qpid/management/authenticationprovider/PrincipalDatabaseAuthenticationManager.js
index bd45d0b488..c5875d59ea 100644
--- a/qpid/java/broker-plugins/management-http/src/main/java/resources/js/qpid/management/authenticationprovider/PrincipalDatabaseAuthenticationManager.js
+++ b/qpid/java/broker-plugins/management-http/src/main/java/resources/js/qpid/management/authenticationprovider/PrincipalDatabaseAuthenticationManager.js
@@ -109,14 +109,14 @@ define(["dojo/_base/xhr",
                      var addUserButton = query(".addUserButton", node)[0];
                      connect.connect(registry.byNode(addUserButton), "onClick", function(evt){ addUser.show(that.name) });
 
-                     var deleteMessagesButton = query(".deleteUserButton", node)[0];
-                                                var deleteWidget = registry.byNode(deleteMessagesButton);
-                                                connect.connect(deleteWidget, "onClick",
-                                                                function(evt){
-                                                                    event.stop(evt);
-                                                                    that.deleteUsers();
-                                                                });
-        }
+                     var deleteUserButton = query(".deleteUserButton", node)[0];
+                     var deleteWidget = registry.byNode(deleteUserButton);
+                     connect.connect(deleteWidget, "onClick",
+                                    function(evt){
+                                        event.stop(evt);
+                                        that.deleteUsers();
+                                    });
+}
 
         DatabaseAuthManager.prototype.deleteUsers = function()
         {
diff --git a/qpid/java/broker-plugins/management-http/src/main/java/resources/js/qpid/management/groupprovider/FileGroupManager.js b/qpid/java/broker-plugins/management-http/src/main/java/resources/js/qpid/management/groupprovider/FileGroupManager.js
index 55d6645a49..c8f991260c 100644
--- a/qpid/java/broker-plugins/management-http/src/main/java/resources/js/qpid/management/groupprovider/FileGroupManager.js
+++ b/qpid/java/broker-plugins/management-http/src/main/java/resources/js/qpid/management/groupprovider/FileGroupManager.js
@@ -125,13 +125,13 @@ define(["dojo/_base/xhr",
                      var addGroupButton = query(".addGroupButton", node)[0];
                      connect.connect(registry.byNode(addGroupButton), "onClick", function(evt){ addGroup.show(groupProviderObj.name) });
 
-                     var deleteMessagesButton = query(".deleteGroupButton", node)[0];
-                                                var deleteWidget = registry.byNode(deleteMessagesButton);
-                                                connect.connect(deleteWidget, "onClick",
-                                                                function(evt){
-                                                                    event.stop(evt);
-                                                                    that.deleteGroups();
-                                                                });
+                     var deleteGroupButton = query(".deleteGroupButton", node)[0];
+                     var deleteWidget = registry.byNode(deleteGroupButton);
+                     connect.connect(deleteWidget, "onClick",
+                                    function(evt){
+                                        event.stop(evt);
+                                        that.deleteGroups();
+                                    });
                 });
         }
 
diff --git a/qpid/java/broker-plugins/management-http/src/main/java/resources/showQueue.html b/qpid/java/broker-plugins/management-http/src/main/java/resources/showQueue.html
index 48a85d68a0..5e6c59eafb 100644
--- a/qpid/java/broker-plugins/management-http/src/main/java/resources/showQueue.html
+++ b/qpid/java/broker-plugins/management-http/src/main/java/resources/showQueue.html
@@ -128,6 +128,7 @@
     <div data-dojo-type="dijit.TitlePane" data-dojo-props="title: 'Messages'">
         <div class="messages"></div>
         <button data-dojo-type="dijit.form.Button" class="deleteMessagesButton" type="button">Delete Messages</button>
+        <button data-dojo-type="dijit.form.Button" class="clearQueueButton" type="button">Clear Queue</button>
         <button data-dojo-type="dijit.form.Button" class="moveMessagesButton" type="button">Move Messages</button>
         <button data-dojo-type="dijit.form.Button" class="copyMessagesButton" type="button">Copy Messages</button>
     </div>
diff --git a/qpid/java/broker/etc/broker_example.acl b/qpid/java/broker/etc/broker_example.acl
index 29dca90f15..418e0d4656 100644
--- a/qpid/java/broker/etc/broker_example.acl
+++ b/qpid/java/broker/etc/broker_example.acl
@@ -83,6 +83,7 @@ ACL ALLOW webadmins ACCESS_LOGS BROKER
 #ACL ALLOW-LOG webadmins UPDATE METHOD component="VirtualHost.Queue" name="moveMessages"
 #ACL ALLOW-LOG webadmins UPDATE METHOD component="VirtualHost.Queue" name="copyMessages"
 #ACL ALLOW-LOG webadmins UPDATE METHOD component="VirtualHost.Queue" name="deleteMessages"
+#ACL ALLOW-LOG webadmins UPDATE METHOD component="VirtualHost.Queue" name="clearQueue"
 
 ### MESSAGING ###
 # The 'ACCESS VIRTUALHOST' rules below apply to messaging operations (as opposed to management operations)
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/systest/rest/MessagesRestTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/systest/rest/MessagesRestTest.java
index 630f0e0c1f..efa4776afd 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/systest/rest/MessagesRestTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/systest/rest/MessagesRestTest.java
@@ -35,8 +35,8 @@ import javax.jms.Destination;
 import javax.jms.Message;
 import javax.jms.MessageProducer;
 import javax.jms.Session;
+import javax.servlet.http.HttpServletResponse;
 
-import org.codehaus.jackson.JsonParseException;
 import org.codehaus.jackson.map.JsonMappingException;
 
 public class MessagesRestTest extends QpidRestTestCase
@@ -161,8 +161,7 @@ public class MessagesRestTest extends QpidRestTestCase
         messagesData.put("destinationQueue", queueName2);
         messagesData.put("move", Boolean.TRUE);
 
-        int status = getRestTestHelper().submitRequest("/service/message/test/" + queueName, "POST", messagesData);
-        assertEquals("Unexpected response code", 200, status);
+        getRestTestHelper().submitRequest("/service/message/test/" + queueName, "POST", messagesData, HttpServletResponse.SC_OK);
 
         // check messages on target queue
         List<Map<String, Object>> messages = getRestTestHelper().getJsonAsList("/service/message/test/" + queueName2);
@@ -213,8 +212,7 @@ public class MessagesRestTest extends QpidRestTestCase
         messagesData.put("messages", copyMessageIds);
         messagesData.put("destinationQueue", queueName2);
 
-        int responseCode = getRestTestHelper().submitRequest("/service/message/test/" + queueName, "POST", messagesData);
-        assertEquals("Unexpected response code", 200, responseCode);
+        getRestTestHelper().submitRequest("/service/message/test/" + queueName, "POST", messagesData, HttpServletResponse.SC_OK);
 
         // check messages on target queue
         List<Map<String, Object>> messages = getRestTestHelper().getJsonAsList("/service/message/test/" + queueName2);
@@ -252,7 +250,7 @@ public class MessagesRestTest extends QpidRestTestCase
         // delete half of the messages
         int deleteNumber = ids.size() / 2;
         StringBuilder queryString = new StringBuilder();
-        List<Long> deleteMessageIds = new ArrayList<Long>();
+        List<Long> deleteMessageIds = new ArrayList<>();
         for (int i = 0; i < deleteNumber; i++)
         {
             Long id = ids.remove(i);
@@ -265,8 +263,7 @@ public class MessagesRestTest extends QpidRestTestCase
         }
 
         // delete messages
-        int responseCode = getRestTestHelper().submitRequest("/service/message/test/" + queueName + "?" + queryString.toString(), "DELETE");
-        assertEquals("Unexpected response code", 200, responseCode);
+        getRestTestHelper().submitRequest("/service/message/test/" + queueName + "?" + queryString.toString(), "DELETE", HttpServletResponse.SC_OK);
 
         // check messages on queue
         List<Map<String, Object>> messages = getRestTestHelper().getJsonAsList("/service/message/test/" + queueName);
@@ -284,7 +281,21 @@ public class MessagesRestTest extends QpidRestTestCase
         }
     }
 
-    private List<Long> getMesssageIds(String queueName) throws IOException, JsonParseException, JsonMappingException
+    public void testClearQueue() throws Exception
+    {
+        String queueName = getTestQueueName();
+
+        // clear queue
+        getRestTestHelper().submitRequest("/service/message/test/" + queueName + "?clear=true", "DELETE", HttpServletResponse.SC_OK);
+
+        // check messages on queue
+        List<Map<String, Object>> messages = getRestTestHelper().getJsonAsList("/service/message/test/" + queueName);
+        assertNotNull("Messages are not found", messages);
+        assertEquals("Unexpected number of messages", 0, messages.size());
+    }
+
+
+    private List<Long> getMesssageIds(String queueName) throws IOException, JsonMappingException
     {
         List<Map<String, Object>> messages = getRestTestHelper().getJsonAsList("/service/message/test/" + queueName);
         List<Long> ids = new ArrayList<Long>();