Move get_nodes from amqqueue to rabbit_amqqueue

We are moving the functionality of getting the nodes/members of an
amqqueue from the `amqqueue` module to `rabbit_amqqueue`. This goes in
the line of previous PRs work towards reducing direct access to the
`QueueTypeState`, such as
#13905. Also, we will
need to discretize different formats of the `nodes` entry in the
`QueueTypeState`, to support both the previous one as a list of nodes
and the new one as a map of nodes to Ra UIds. Doing so in a module such
as `amqqueue`, which feels like an accessor module around the `amqqueue`
record, doesn't feel right.

Author: LoisSotoLopez

deps/rabbit/src/amqqueue.erl

@@ -29,7 +29,6 @@
          % exclusive_owner
          get_exclusive_owner/1,
          get_leader_node/1,
-         get_nodes/1,
          % name (#resource)
          get_name/1,
          set_name/2,
@@ -394,15 +393,6 @@ get_leader_node(#amqqueue{pid = {_, Leader}}) -> Leader;
 get_leader_node(#amqqueue{pid = none}) -> none;
 get_leader_node(#amqqueue{pid = Pid}) -> node(Pid).
 
--spec get_nodes(amqqueue_v2()) -> [node(),...].
-
-get_nodes(Q) ->
-    case amqqueue:get_type_state(Q) of
-        #{nodes := Nodes} ->
-            Nodes;
-        _ ->
-            [get_leader_node(Q)]
-    end.
 
 % operator_policy
 

deps/rabbit/src/rabbit_amqp_management.erl

@@ -465,7 +465,7 @@ encode_queue(Q, NumMsgs, NumConsumers) ->
     {Leader :: node() | none, Replicas :: [node(),...]}.
 queue_topology(Q) ->
     Leader = amqqueue:get_leader_node(Q),
-    Replicas = amqqueue:get_nodes(Q),
+    Replicas = rabbit_amqqueue:get_nodes(Q),
     {Leader, Replicas}.
 
 decode_exchange({map, KVList}) ->

deps/rabbit/src/rabbit_amqqueue.erl

@@ -23,6 +23,7 @@
 -export([list/0, list_durable/0, list/1, info_keys/0, info/1, info/2, info_all/1, info_all/2,
          emit_info_all/5, list_local/1, info_local/1,
          emit_info_local/4, emit_info_down/4]).
+-export([get_nodes/1]).
 -export([count/0]).
 -export([list_down/1, list_down/2, list_all/1,
          count/1, list_names/0, list_names/1, list_local_names/0,
@@ -1234,6 +1235,12 @@ list() ->
 count() ->
     rabbit_db_queue:count().
 
+-spec get_nodes(amqqueue:amqqueue_v2()) -> [node(),...].
+
+get_nodes(Q) ->
+    [{members, Nodes}] = info(Q, [members]),
+    Nodes.
+
 -spec list_names() -> [name()].
 
 list_names() ->
@@ -2043,12 +2050,7 @@ pseudo_queue(#resource{kind = queue} = QueueName, Pid, Durable)
                 ).
 
 get_quorum_nodes(Q) ->
-    case amqqueue:get_type_state(Q) of
-        #{nodes := Nodes} ->
-            Nodes;
-        _ ->
-            []
-    end.
+    rabbit_amqqueue:get_nodes(Q).
 
 -spec prepend_extra_bcc(Qs) ->
     Qs when Qs :: [amqqueue:amqqueue() |

deps/rabbit/src/rabbit_queue_location.erl

@@ -143,7 +143,7 @@ select_members(Size, _, AllNodes, RunningNodes, _, _, GetQueues) ->
     Counters0 = maps:from_list([{N, 0} || N <- lists:delete(?MODULE:node(), AllNodes)]),
     Queues = GetQueues(),
     Counters = lists:foldl(fun(Q, Acc) ->
-                                   #{nodes := Nodes} = amqqueue:get_type_state(Q),
+                                   Nodes = rabbit_amqqueue:get_nodes(Q),
                                    lists:foldl(fun(N, A)
                                                      when is_map_key(N, A) ->
                                                        maps:update_with(N, fun(C) -> C+1 end, A);

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -2178,7 +2178,7 @@ force_checkpoint_on_queue(QName) ->
         {ok, Q} when ?amqqueue_is_quorum(Q) ->
             {RaName, _} = amqqueue:get_pid(Q),
             ?LOG_DEBUG("Sending command to force ~ts to take a checkpoint", [QNameFmt]),
-            Nodes = amqqueue:get_nodes(Q),
+            Nodes = rabbit_amqqueue:get_nodes(Q),
             _ = [ra:cast_aux_command({RaName, Node}, force_checkpoint)
                  || Node <- Nodes],
             ok;

deps/rabbit/src/rabbit_stream_coordinator.erl

@@ -153,8 +153,8 @@ stop() ->
 
 new_stream(Q, LeaderNode)
   when ?is_amqqueue(Q) andalso is_atom(LeaderNode) ->
-    #{name := StreamId,
-      nodes := Nodes} = amqqueue:get_type_state(Q),
+    #{name := StreamId} = amqqueue:get_type_state(Q),
+    Nodes = rabbit_amqqueue:get_nodes(Q),
     %% assertion leader is in nodes configuration
     true = lists:member(LeaderNode, Nodes),
     process_command({new_stream, StreamId,

deps/rabbitmq_ct_helpers/src/queue_utils.erl

@@ -208,7 +208,7 @@ assert_number_of_replicas(Config, Server, VHost, QQ, Count) ->
        begin
            {ok, Q} = rabbit_ct_broker_helpers:rpc(
                        Config, Server, rabbit_amqqueue, lookup, [QQ, VHost]),
-           #{nodes := Nodes} = amqqueue:get_type_state(Q),
+           Nodes = rabbit_amqqueue:get_nodes(Q),
            length(Nodes)
        end,
        30000).

deps/rabbitmq_prometheus/src/collectors/prometheus_rabbitmq_core_metrics_collector.erl

@@ -451,8 +451,7 @@ emit_queue_info(Prefix, VHostsFilter, Callback) ->
                            true -> Acc;
                            false ->
                                Type = amqqueue:get_type(Q),
-                               TypeState = amqqueue:get_type_state(Q),
-                               Members = maps:get(nodes, TypeState, []),
+                               Members = rabbit_amqqueue:get_nodes(Q),
                                case membership(amqqueue:get_pid(Q), Members) of
                                    not_a_member ->
                                        Acc;