Merge pull request #1801 from rabbitmq/qq-list-consumers

Implement consumer listing for quorum queues

Author: acogoluegnes

src/lqueue.erl

@@ -25,27 +25,32 @@
 
 -define(QUEUE, queue).
 
--export_type([?MODULE/0]).
+-export_type([
+              ?MODULE/0,
+              ?MODULE/1
+             ]).
 
--opaque ?MODULE() :: {non_neg_integer(), queue:queue()}.
+-opaque ?MODULE() :: {non_neg_integer(), queue:queue(term())}.
+-opaque ?MODULE(T) :: {non_neg_integer(), queue:queue(T)}.
 -type value()     :: any().
--type result()    :: 'empty' | {'value', value()}.
+-type result(T)    :: 'empty' | {'value', T}.
 
--spec new() -> ?MODULE().
--spec drop(?MODULE()) -> ?MODULE().
--spec is_empty(?MODULE()) -> boolean().
--spec len(?MODULE()) -> non_neg_integer().
--spec in(value(), ?MODULE()) -> ?MODULE().
+-spec new() -> ?MODULE(_).
+-spec drop(?MODULE(T)) -> ?MODULE(T).
+-spec is_empty(?MODULE(_)) -> boolean().
+-spec len(?MODULE(_)) -> non_neg_integer().
+-spec in(T, ?MODULE(T)) -> ?MODULE(T).
 -spec in_r(value(), ?MODULE()) -> ?MODULE().
--spec out(?MODULE()) -> {result(), ?MODULE()}.
--spec out_r(?MODULE()) -> {result(), ?MODULE()}.
--spec join(?MODULE(), ?MODULE()) -> ?MODULE().
--spec foldl(fun ((value(), B) -> B), B, ?MODULE()) -> B.
--spec foldr(fun ((value(), B) -> B), B, ?MODULE()) -> B.
--spec from_list([value()]) -> ?MODULE().
--spec to_list(?MODULE()) -> [value()].
--spec peek(?MODULE()) -> result().
--spec peek_r(?MODULE()) -> result().
+-spec out(?MODULE(T)) -> {result(T), ?MODULE()}.
+-spec out_r(?MODULE(T)) -> {result(T), ?MODULE()}.
+-spec join(?MODULE(A), ?MODULE(B)) -> ?MODULE(A | B).
+-spec foldl(fun ((T, B) -> B), B, ?MODULE(T)) -> B.
+-spec foldr(fun ((T, B) -> B), B, ?MODULE(T)) -> B.
+-spec from_list([T]) -> ?MODULE(T).
+-spec to_list(?MODULE(T)) -> [T].
+% -spec peek(?MODULE()) -> result().
+-spec peek(?MODULE(T)) -> result(T).
+-spec peek_r(?MODULE(T)) -> result(T).
 
 new() -> {0, ?QUEUE:new()}.
 

src/rabbit_amqqueue.erl

@@ -919,8 +919,12 @@ notify_policy_changed(#amqqueue{pid = QPid,
                                 name = QName}) when ?IS_QUORUM(QPid) ->
     rabbit_quorum_queue:policy_changed(QName, QPid).
 
-consumers(#amqqueue{ pid = QPid }) ->
-    delegate:invoke(QPid, {gen_server2, call, [consumers, infinity]}).
+consumers(#amqqueue{pid = QPid}) when ?IS_CLASSIC(QPid) ->
+    delegate:invoke(QPid, {gen_server2, call, [consumers, infinity]});
+consumers(#amqqueue{pid = QPid}) when ?IS_QUORUM(QPid) ->
+    {ok, {_, Result}, _} = ra:local_query(QPid,
+                                          fun rabbit_fifo:query_consumers/1),
+    maps:values(Result).
 
 consumer_info_keys() -> ?CONSUMER_INFO_KEYS.
 
@@ -1147,14 +1151,15 @@ basic_get(#amqqueue{pid = {Name, _} = Id, type = quorum, name = QName} = Q, _ChP
                                        [rabbit_misc:rs(QName), Reason])
     end.
 
-basic_consume(#amqqueue{pid = QPid, name = QName, type = classic}, NoAck, ChPid, LimiterPid,
-              LimiterActive, ConsumerPrefetchCount, ConsumerTag,
+basic_consume(#amqqueue{pid = QPid, name = QName, type = classic}, NoAck, ChPid,
+              LimiterPid, LimiterActive, ConsumerPrefetchCount, ConsumerTag,
               ExclusiveConsume, Args, OkMsg, ActingUser, QState) ->
     ok = check_consume_arguments(QName, Args),
-    case delegate:invoke(QPid, {gen_server2, call,
-                                [{basic_consume, NoAck, ChPid, LimiterPid, LimiterActive,
-                                  ConsumerPrefetchCount, ConsumerTag, ExclusiveConsume,
-                                  Args, OkMsg, ActingUser}, infinity]}) of
+    case delegate:invoke(QPid,
+                         {gen_server2, call,
+                          [{basic_consume, NoAck, ChPid, LimiterPid, LimiterActive,
+                            ConsumerPrefetchCount, ConsumerTag, ExclusiveConsume,
+                            Args, OkMsg, ActingUser}, infinity]}) of
         ok ->
             {ok, QState};
         Err ->
@@ -1164,15 +1169,17 @@ basic_consume(#amqqueue{type = quorum}, _NoAck, _ChPid,
               _LimiterPid, true, _ConsumerPrefetchCount, _ConsumerTag,
               _ExclusiveConsume, _Args, _OkMsg, _ActingUser, _QStates) ->
     {error, global_qos_not_supported_for_queue_type};
-basic_consume(#amqqueue{pid = {Name, _} = Id, name = QName, type = quorum} = Q, NoAck, ChPid,
-              _LimiterPid, _LimiterActive, ConsumerPrefetchCount, ConsumerTag,
-              ExclusiveConsume, Args, OkMsg, _ActingUser, QStates) ->
+basic_consume(#amqqueue{pid = {Name, _} = Id, name = QName, type = quorum} = Q,
+              NoAck, ChPid, _LimiterPid, _LimiterActive, ConsumerPrefetchCount,
+              ConsumerTag, ExclusiveConsume, Args, OkMsg,
+              ActingUser, QStates) ->
     ok = check_consume_arguments(QName, Args),
     QState0 = get_quorum_state(Id, QName, QStates),
     {ok, QState} = rabbit_quorum_queue:basic_consume(Q, NoAck, ChPid,
                                                      ConsumerPrefetchCount,
                                                      ConsumerTag,
                                                      ExclusiveConsume, Args,
+                                                     ActingUser,
                                                      OkMsg, QState0),
     {ok, maps:put(Name, QState, QStates)}.