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

@@ -464,7 +464,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,
@@ -1233,6 +1234,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() ->
@@ -2042,12 +2049,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_jms_selector_parser.erl

@@ -1,6 +1,4 @@
--file("rabbit_jms_selector_parser.yrl", 0).
 -module(rabbit_jms_selector_parser).
--file("rabbit_jms_selector_parser.erl", 3).
 -export([parse/1, parse_and_scan/1, format_error/1]).
 -file("rabbit_jms_selector_parser.yrl", 122).
 
@@ -26,9 +24,7 @@ process_escape_char({string, Line, Value}) ->
 %%
 %% %CopyrightBegin%
 %%
-%% SPDX-License-Identifier: Apache-2.0
-%%
-%% Copyright Ericsson AB 1996-2025. All Rights Reserved.
+%% Copyright Ericsson AB 1996-2021. All Rights Reserved.
 %%
 %% Licensed under the Apache License, Version 2.0 (the "License");
 %% you may not use this file except in compliance with the License.
@@ -50,16 +46,10 @@ process_escape_char({string, Line, Value}) ->
 
 -type yecc_ret() :: {'error', _} | {'ok', _}.
 
--ifdef (YECC_PARSE_DOC).
--doc ?YECC_PARSE_DOC.
--endif.
 -spec parse(Tokens :: list()) -> yecc_ret().
 parse(Tokens) ->
     yeccpars0(Tokens, {no_func, no_location}, 0, [], []).
 
--ifdef (YECC_PARSE_AND_SCAN_DOC).
--doc ?YECC_PARSE_AND_SCAN_DOC.
--endif.
 -spec parse_and_scan({function() | {atom(), atom()}, [_]}
                      | {atom(), atom(), [_]}) -> yecc_ret().
 parse_and_scan({F, A}) ->
@@ -68,9 +58,6 @@ parse_and_scan({M, F, A}) ->
     Arity = length(A),
     yeccpars0([], {{fun M:F/Arity, A}, no_location}, 0, [], []).
 
--ifdef (YECC_FORMAT_ERROR_DOC).
--doc ?YECC_FORMAT_ERROR_DOC.
--endif.
 -spec format_error(any()) -> [char() | list()].
 format_error(Message) ->
     case io_lib:deep_char_list(Message) of
@@ -212,7 +199,7 @@ yecctoken2string1(Other) ->
 
 
 
--file("rabbit_jms_selector_parser.erl", 215).
+-file("rabbit_jms_selector_parser.erl", 202).
 
 -dialyzer({nowarn_function, yeccpars2/7}).
 -compile({nowarn_unused_function,  yeccpars2/7}).

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

@@ -2176,7 +2176,7 @@ force_checkpoint_on_queue(QName) ->
         {ok, Q} when ?amqqueue_is_quorum(Q) ->
             {RaName, _} = amqqueue:get_pid(Q),
             rabbit_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

@@ -152,8 +152,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/rabbit/test/quorum_queue_SUITE.erl

@@ -1310,7 +1310,7 @@ force_shrink_member_to_current_member(Config) ->
             wait_for_messages_ready([Server0], RaName, 3),
 
             {ok, Q0} = rpc:call(Server0, rabbit_amqqueue, lookup, [QQ, <<"/">>]),
-            #{nodes := Nodes0} = amqqueue:get_type_state(Q0),
+            Nodes0 = rabbit_amqqueue:get_nodes(Q0),
             ?assertEqual(3, length(Nodes0)),
 
             rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue,
@@ -1319,15 +1319,15 @@ force_shrink_member_to_current_member(Config) ->
             wait_for_messages_ready([Server0], RaName, 3),
 
             {ok, Q1} = rpc:call(Server0, rabbit_amqqueue, lookup, [QQ, <<"/">>]),
-            #{nodes := Nodes1} = amqqueue:get_type_state(Q1),
+            Nodes1 = rabbit_amqqueue:get_nodes(Q1),
             ?assertEqual(1, length(Nodes1)),
 
             %% grow queues back to all nodes
             [rpc:call(Server0, rabbit_quorum_queue, grow, [S, <<"/">>, <<".*">>, all]) || S <- [Server1, Server2]],
 
             wait_for_messages_ready([Server0], RaName, 3),
             {ok, Q2} = rpc:call(Server0, rabbit_amqqueue, lookup, [QQ, <<"/">>]),
-            #{nodes := Nodes2} = amqqueue:get_type_state(Q2),
+            Nodes2 = rabbit_amqqueue:get_nodes(Q2),
             ?assertEqual(3, length(Nodes2))
     end.
 
@@ -1354,7 +1354,7 @@ force_all_queues_shrink_member_to_current_member(Config) ->
                  rabbit_ct_client_helpers:publish(Ch, Q, 3),
                  wait_for_messages_ready([Server0], RaName, 3),
                  {ok, Q0} = rpc:call(Server0, rabbit_amqqueue, lookup, [Q, <<"/">>]),
-                 #{nodes := Nodes0} = amqqueue:get_type_state(Q0),
+                 Nodes0 = rabbit_amqqueue:get_nodes(Q0),
                  ?assertEqual(3, length(Nodes0))
              end || Q <- QQs],
 
@@ -1365,7 +1365,7 @@ force_all_queues_shrink_member_to_current_member(Config) ->
                  RaName = ra_name(Q),
                  wait_for_messages_ready([Server0], RaName, 3),
                  {ok, Q0} = rpc:call(Server0, rabbit_amqqueue, lookup, [Q, <<"/">>]),
-                 #{nodes := Nodes0} = amqqueue:get_type_state(Q0),
+                 Nodes0 = rabbit_amqqueue:get_nodes(Q0),
                  ?assertEqual(1, length(Nodes0))
              end || Q <- QQs],
 
@@ -1376,7 +1376,7 @@ force_all_queues_shrink_member_to_current_member(Config) ->
                  RaName = ra_name(Q),
                  wait_for_messages_ready([Server0], RaName, 3),
                  {ok, Q0} = rpc:call(Server0, rabbit_amqqueue, lookup, [Q, <<"/">>]),
-                 #{nodes := Nodes0} = amqqueue:get_type_state(Q0),
+                 Nodes0 = rabbit_amqqueue:get_nodes(Q0),
                  ?assertEqual(3, length(Nodes0))
              end || Q <- QQs]
     end.
@@ -1420,7 +1420,7 @@ force_vhost_queues_shrink_member_to_current_member(Config) ->
                 {ok, RaName} = rpc:call(Server0, rabbit_queue_type_util, qname_to_internal_name, [QQRes]),
                 wait_for_messages_ready([Server0], RaName, 3),
                 {ok, Q0} = rpc:call(Server0, rabbit_amqqueue, lookup, [Q, VHost]),
-                #{nodes := Nodes0} = amqqueue:get_type_state(Q0),
+                Nodes0 = rabbit_amqqueue:get_nodes(Q0),
                 ?assertEqual(3, length(Nodes0))
             end || Q <- QQs, VHost <- VHosts],
 
@@ -1432,7 +1432,7 @@ force_vhost_queues_shrink_member_to_current_member(Config) ->
                 {ok, RaName} = rpc:call(Server0, rabbit_queue_type_util, qname_to_internal_name, [QQRes]),
                 wait_for_messages_ready([Server0], RaName, 3),
                 {ok, Q0} = rpc:call(Server0, rabbit_amqqueue, lookup, [Q, VHost]),
-                #{nodes := Nodes0} = amqqueue:get_type_state(Q0),
+                Nodes0 = rabbit_amqqueue:get_nodes(Q0),
                 case VHost of
                     VHost1 -> ?assertEqual(3, length(Nodes0));
                     VHost2 -> ?assertEqual(1, length(Nodes0))
@@ -1447,7 +1447,7 @@ force_vhost_queues_shrink_member_to_current_member(Config) ->
                 {ok, RaName} = rpc:call(Server0, rabbit_queue_type_util, qname_to_internal_name, [QQRes]),
                 wait_for_messages_ready([Server0], RaName, 3),
                 {ok, Q0} = rpc:call(Server0, rabbit_amqqueue, lookup, [Q, VHost]),
-                #{nodes := Nodes0} = amqqueue:get_type_state(Q0),
+                Nodes0 = rabbit_amqqueue:get_nodes(Q0),
                 ?assertEqual(3, length(Nodes0))
             end || Q <- QQs, VHost <- VHosts]
     end.
@@ -2955,7 +2955,7 @@ delete_member_member_already_deleted(Config) ->
                  rpc:call(Server, rabbit_quorum_queue, delete_member,
                           [<<"/">>, QQ, Server2])),
     {ok, Q} = rpc:call(Server, rabbit_amqqueue, lookup, [QQ, <<"/">>]),
-    #{nodes := Nodes} = amqqueue:get_type_state(Q),
+    Nodes = rabbit_amqqueue:get_nodes(Q),
     ?assertEqual(1, length(Nodes)),
     ok.
 

deps/rabbitmq_prometheus/src/collectors/prometheus_rabbitmq_core_metrics_collector.erl

@@ -457,8 +457,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;