Queues with plugins - remove queue_topology callback

Author: ikavgo

deps/rabbit/src/amqqueue.erl

@@ -28,7 +28,8 @@
          set_decorators/2,
          % exclusive_owner
          get_exclusive_owner/1,
-         get_leader/1,
+         get_leader_node/1,
+         get_nodes/1,
          % name (#resource)
          get_name/1,
          set_name/2,
@@ -387,10 +388,21 @@ set_decorators(#amqqueue{} = Queue, Decorators) ->
 get_exclusive_owner(#amqqueue{exclusive_owner = Owner}) ->
     Owner.
 
--spec get_leader(amqqueue_v2()) -> node().
+-spec get_leader_node(amqqueue_v2()) -> node() | none.
 
-%% TODO: not only qqs can have leaders, dispatch via queue type
-get_leader(#amqqueue{type = rabbit_quorum_queue, pid = {_, Leader}}) -> Leader.
+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

@@ -463,8 +463,9 @@ encode_queue(Q, NumMsgs, NumConsumers) ->
 -spec queue_topology(amqqueue:amqqueue()) ->
     {Leader :: node() | none, Replicas :: [node(),...]}.
 queue_topology(Q) ->
-    Type = amqqueue:get_type(Q),
-    Type:queue_topology(Q).
+    Leader = amqqueue:get_leader_node(Q),
+    Replicas = amqqueue:get_nodes(Q),
+    {Leader, Replicas}.
 
 decode_exchange({map, KVList}) ->
     M = lists:foldl(

deps/rabbit/src/rabbit_amqqueue.erl

@@ -1325,14 +1325,14 @@ list_stream_queues_on(Node) when is_atom(Node) ->
 list_local_leaders() ->
     [ Q || Q <- list(),
          amqqueue:is_quorum(Q),
-         amqqueue:get_state(Q) =/= crashed, amqqueue:get_leader(Q) =:= node()].
+         amqqueue:get_state(Q) =/= crashed, amqqueue:get_leader_node(Q) =:= node()].
 
 -spec list_local_followers() -> [amqqueue:amqqueue()].
 list_local_followers() ->
     [Q
       || Q <- list(),
          amqqueue:is_quorum(Q),
-         amqqueue:get_leader(Q) =/= node(),
+         amqqueue:get_leader_node(Q) =/= node(),
          lists:member(node(), get_quorum_nodes(Q)),
          rabbit_quorum_queue:is_recoverable(Q)
          ].

deps/rabbit/src/rabbit_classic_queue.erl

@@ -64,8 +64,7 @@
          send_drained_credit_api_v1/4,
          send_credit_reply/7]).
 
--export([queue_topology/1,
-         policy_apply_to_name/0,
+-export([policy_apply_to_name/0,
          stop/1,
          list_with_minimum_quorum/0,
          drain/1,
@@ -702,13 +701,6 @@ send_credit_reply(Pid, QName, Ctag, DeliveryCount, Credit, Available, Drain) ->
 send_queue_event(Pid, QName, Event) ->
     gen_server:cast(Pid, {queue_event, QName, Event}).
 
--spec queue_topology(amqqueue:amqqueue()) ->
-    {Leader :: node() | none, Replicas :: [node(),...]}.
-queue_topology(Q) ->
-    Pid = amqqueue:get_pid(Q),
-    Node = node(Pid),
-    {Node, [Node]}.
-
 policy_apply_to_name() ->
     <<"classic_queues">>.
 

deps/rabbit/src/rabbit_queue_type.erl

@@ -284,9 +284,6 @@
 -callback notify_decorators(amqqueue:amqqueue()) ->
     ok.
 
--callback queue_topology(amqqueue:amqqueue()) ->
-    {Leader :: undefined | node(), Replicas :: undefined | [node(),...]}.
-
 -callback policy_apply_to_name() -> binary().
 
 %% -callback on_node_up(node()) -> ok.

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -77,8 +77,7 @@
          force_vhost_queues_shrink_member_to_current_member/1,
          force_all_queues_shrink_member_to_current_member/0]).
 
--export([queue_topology/1,
-         policy_apply_to_name/0,
+-export([policy_apply_to_name/0,
          drain/1,
          revive/0,
          queue_vm_stats_sups/0,
@@ -2253,25 +2252,6 @@ maybe_log_leader_health_check_result(Result) ->
     Qs = lists:map(fun(R) -> catch maps:get(<<"readable_name">>, R) end, Result),
     rabbit_log:warning("Leader health check result (unhealthy leaders detected): ~tp", [Qs]).
 
--spec queue_topology(amqqueue:amqqueue()) ->
-          {Leader :: node() | none, Replicas :: [node(),...]}.
-queue_topology(Q) ->
-    Leader = case amqqueue:get_pid(Q) of
-                 {_RaName, Node} ->
-                     Node;
-                 none ->
-                     none;
-                 Pid ->
-                     node(Pid)
-             end,
-    Replicas = case amqqueue:get_type_state(Q) of
-                   #{nodes := Nodes} ->
-                       Nodes;
-                   _ ->
-                       [Leader]
-               end,
-    {Leader, Replicas}.
-
 policy_apply_to_name() ->
     <<"quorum_queues">>.
 

deps/rabbit/src/rabbit_stream_queue.erl

@@ -59,8 +59,7 @@
 
 -export([check_max_segment_size_bytes/1]).
 
--export([queue_topology/1,
-         policy_apply_to_name/0,
+-export([policy_apply_to_name/0,
          stop/1,
          drain/1,
          revive/0,
@@ -1430,25 +1429,6 @@ delivery_count_add(none, _) ->
 delivery_count_add(Count, N) ->
     serial_number:add(Count, N).
 
--spec queue_topology(amqqueue:amqqueue()) ->
-    {Leader :: node() | none, Replicas :: [node(),...]}.
-queue_topology(Q) ->
-    Leader = case amqqueue:get_pid(Q) of
-                 {_RaName, Node} ->
-                     Node;
-                 none ->
-                     none;
-                 Pid ->
-                     node(Pid)
-             end,
-    Replicas = case amqqueue:get_type_state(Q) of
-                   #{nodes := Nodes} ->
-                       Nodes;
-                   _ ->
-                       [Leader]
-               end,
-    {Leader, Replicas}.
-
 policy_apply_to_name() ->
     <<"streams">>.