add 'rabbitmqctl list_consumers' command

2 files changed, 21 insertions, 0 deletions

diff --git a/docs/rabbitmqctl.1.pod b/docs/rabbitmqctl.1.pod
index b82d5b87..0ebe17dc 100644
--- a/docs/rabbitmqctl.1.pod
+++ b/docs/rabbitmqctl.1.pod
@@ -422,6 +422,14 @@ QoS prefetch count limit in force, 0 if unlimited
 
 =back
 
+=item list_consumers
+
+List consumers, i.e. subscriptions to a queue's message stream. Each
+line printed shows, separated by tab characters, the id of the channel
+process via which the subscription was created and is managed, the
+consumer tag which uniquely identifies the subscription within a
+channel, and the name of the queue subscribed to.
+
 The list_queues, list_exchanges and list_bindings commands accept an
 optional virtual host parameter for which to display results,
 defaulting to I<"/">. The default can be overridden with the B<-p>
diff --git a/src/rabbit_control.erl b/src/rabbit_control.erl
index 738ed444..232b3c8f 100644
--- a/src/rabbit_control.erl
+++ b/src/rabbit_control.erl
@@ -159,6 +159,7 @@ Available commands:
   list_bindings  [-p <VHostPath>]
   list_connections [<ConnectionInfoItem> ...]
   list_channels [<ChannelInfoItem> ...]
+  list_consumers
 
 Quiet output mode is selected with the \"-q\" flag. Informational
 messages are suppressed when quiet mode is in effect.
@@ -197,6 +198,9 @@ vhost, transactional, consumer_count, messages_unacknowledged,
 prefetch_count]. The default is to display pid, user, transactional,
 consumer_count, messages_unacknowledged.
 
+The output format for \"list_consumers\" is a list of rows containing
+the channel process id, consumer tag and queue name, in that order.
+
 "),
     halt(1).
 
@@ -314,6 +318,15 @@ action(list_channels, Node, Args, Inform) ->
     display_info_list(rpc_call(Node, rabbit_channel, info_all, [ArgAtoms]),
                       ArgAtoms);
 
+action(list_consumers, Node, [], Inform) ->
+    Inform("Listing consumers", []),
+    InfoKeys = [channel_pid, consumer_tag, queue_name],
+    display_info_list(
+      [lists:zip(InfoKeys, tuple_to_list(X)) ||
+          X <- rpc_call(Node, rabbit_channel, consumers_all, [])],
+      InfoKeys),
+    ok;
+
 action(Command, Node, Args, Inform) ->
     {VHost, RemainingArgs} = parse_vhost_flag(Args),
     action(Command, Node, VHost, RemainingArgs, Inform).