QPID-1214 - Committed William's patch

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

3 files changed, 25 insertions, 7 deletions

diff --git a/cpp/src/qpid/broker/Queue.cpp b/cpp/src/qpid/broker/Queue.cpp
index 18a3574cdd..ebb143a472 100644
--- a/cpp/src/qpid/broker/Queue.cpp
+++ b/cpp/src/qpid/broker/Queue.cpp
@@ -31,6 +31,7 @@
 #include "qpid/framing/reply_exceptions.h"
 #include "qpid/sys/Monitor.h"
 #include "qpid/sys/Time.h"
+#include "qpid/management/ArgsQueuePurge.h"
 
 #include <iostream>
 #include <algorithm>
@@ -372,11 +373,24 @@ QueuedMessage Queue::get(){
     return msg;
 }
 
-uint32_t Queue::purge(){
+/**
+ * purge - for purging all or some messages on a queue
+ *         depending on the purge_request
+ *
+ * purge_request == 0 then purge all messages
+ *               == N then purge N messages from queue
+ * Sometimes purge_request == 1 to unblock the top of queue
+ */
+uint32_t Queue::purge(const uint32_t purge_request){
     Mutex::ScopedLock locker(messageLock);
-    int count = messages.size();
-    while(!messages.empty()) {
+    uint32_t purge_count = purge_request; // only comes into play if  >0 
+
+    uint32_t count = 0;
+    // Either purge them all or just the some (purge_count) while the queue isn't empty.
+    while((!purge_request || purge_count--) && !messages.empty()) 
+    {
         popAndDequeue();
+	count++;
     }
     return count;
 }
@@ -710,7 +724,7 @@ ManagementObject* Queue::GetManagementObject (void) const
 }
 
 Manageable::status_t Queue::ManagementMethod (uint32_t methodId,
-                                              Args&    /*args*/)
+                                              Args&    args)
 {
     Manageable::status_t status = Manageable::STATUS_UNKNOWN_METHOD;
 
@@ -719,7 +733,8 @@ Manageable::status_t Queue::ManagementMethod (uint32_t methodId,
     switch (methodId)
     {
     case management::Queue::METHOD_PURGE :
-        purge ();
+      management::ArgsQueuePurge iargs = dynamic_cast<const management::ArgsQueuePurge&>(args);
+        purge (iargs.i_request);
         status = Manageable::STATUS_OK;
         break;
     }
diff --git a/cpp/src/qpid/broker/Queue.h b/cpp/src/qpid/broker/Queue.h
index f1694eb5a4..2d238ff57d 100644
--- a/cpp/src/qpid/broker/Queue.h
+++ b/cpp/src/qpid/broker/Queue.h
@@ -161,7 +161,8 @@ namespace qpid {
             void consume(Consumer& c, bool exclusive = false);
             void cancel(Consumer& c);
 
-            uint32_t purge();
+            uint32_t purge(const uint32_t purge_request = 0); //defaults to all messages 
+
             uint32_t getMessageCount() const;
             uint32_t getConsumerCount() const;
             inline const string& getName() const { return name; }
diff --git a/specs/management-schema.xml b/specs/management-schema.xml
index 23eda40df2..850f9c62e6 100644
--- a/specs/management-schema.xml
+++ b/specs/management-schema.xml
@@ -160,7 +160,9 @@
     <statistic name="unackedMessages"     type="hilo32"   unit="message"     desc="Messages consumed but not yet acked"/>
     <statistic name="messageLatency"      type="mmaTime"  unit="nanosecond"  desc="Broker latency through this queue"/>
 
-    <method name="purge" desc="Discard all messages on queue"/>
+    <method name="purge" desc="Discard all or some messages on a queue">
+      <arg name="request"          dir="I" type="uint32" desc="0 for all messages or n>0 for n messages"/>
+    </method>
   </class>
 
   <!--