Introduce rabbitmqctl purge_queue

For operational convenience.

docs/rabbitmqctl.1.xml

@@ -589,6 +589,25 @@
             </para>
           </listitem>
         </varlistentry>
+        <varlistentry>
+          <term><cmdsynopsis><command>purge_queue</command> <arg choice="req">queue</arg></cmdsynopsis>
+          </term>
+          <listitem>
+            <variablelist>
+              <varlistentry>
+                <term>queue</term>
+                <listitem>
+                  <para>
+                    The name of the queue to purge.
+                  </para>
+                </listitem>
+              </varlistentry>
+            </variablelist>
+            <para>
+              Purges a queue (removes all messages in it).
+            </para>
+          </listitem>
+        </varlistentry>
         <varlistentry>
           <term><cmdsynopsis><command>set_cluster_name</command> <arg choice="req">name</arg></cmdsynopsis></term>
           <listitem>

src/rabbit_control_main.erl

@@ -19,7 +19,8 @@
 -include("rabbit_cli.hrl").
 
 -export([start/0, stop/0, parse_arguments/2, action/5,
-         sync_queue/1, cancel_sync_queue/1, become/1]).
+         sync_queue/1, cancel_sync_queue/1, become/1,
+         purge_queue/1]).
 
 -import(rabbit_cli, [rpc_call/4, rpc_call/5]).
 
@@ -45,6 +46,7 @@
          cluster_status,
          {sync_queue, [?VHOST_DEF]},
          {cancel_sync_queue, [?VHOST_DEF]},
+         {purge_queue, [?VHOST_DEF]},
 
          add_user,
          delete_user,
@@ -111,7 +113,8 @@
 -define(COMMANDS_WITH_TIMEOUT,
         [list_user_permissions, list_policies, list_queues, list_exchanges,
         list_bindings, list_connections, list_channels, list_consumers,
-        list_vhosts, list_parameters]).
+        list_vhosts, list_parameters,
+        purge_queue]).
 
 %%----------------------------------------------------------------------------
 
@@ -485,6 +488,15 @@ action(Command, Node, Args, Opts, Inform) ->
     %% the default timeout.
     action(Command, Node, Args, Opts, Inform, ?RPC_TIMEOUT).
 
+action(purge_queue, Node, [], Opts, Inform, Timeout) ->
+    {error, "purge_queue takes queue name as an argument"};
+
+action(purge_queue, Node, [Q], Opts, Inform, Timeout) ->
+    VHost = proplists:get_value(?VHOST_OPT, Opts),
+    QRes = rabbit_misc:r(list_to_binary(VHost), queue, list_to_binary(Q)),
+    Inform("Purging ~s", [rabbit_misc:rs(QRes)]),
+    rpc_call(Node, rabbit_control_main, purge_queue, [QRes], Timeout);
+
 action(list_users, Node, [], _Opts, Inform, Timeout) ->
     Inform("Listing users", []),
     display_info_list(
@@ -589,6 +601,13 @@ cancel_sync_queue(Q) ->
                  rabbit_amqqueue:cancel_sync_mirrors(QPid)
          end).
 
+purge_queue(Q) ->
+    rabbit_amqqueue:with(
+      Q, fun(Q1) ->
+                 rabbit_amqqueue:purge(Q1),
+                 ok
+         end).
+
 %%----------------------------------------------------------------------------
 
 wait_for_application(Node, PidFile, Application, Inform) ->