Add queue_topology/1 callback to rabbit_queue_type

Co-authored-by: ikavgo <[email protected]>

Author: mkuratczyk

deps/rabbit/src/rabbit_classic_queue.erl

@@ -66,6 +66,8 @@
 
 -export([validate_policy/1]).
 
+-export([queue_topology/1]).
+
 -rabbit_boot_step(
    {?MODULE,
     [{description, "Deprecated queue-master-locator support."
@@ -674,3 +676,10 @@ 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 :: undefined | node(), Replicas :: undefined | [node(),...]}.
+queue_topology(Q) ->
+    Pid = amqqueue:get_pid(Q),
+    Node = node(Pid),
+    {Node, [Node]}.

deps/rabbit/src/rabbit_queue_type.erl

@@ -274,6 +274,9 @@
 -callback notify_decorators(amqqueue:amqqueue()) ->
     ok.
 
+-callback queue_topology(amqqueue:amqqueue()) ->
+    {Leader :: undefined | node(), Replicas :: undefined | [node(),...]}.
+
 -spec discover(binary() | atom()) -> queue_type().
 discover(<<"undefined">>) ->
     fallback();

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -77,6 +77,8 @@
          force_vhost_queues_shrink_member_to_current_member/1,
          force_all_queues_shrink_member_to_current_member/0]).
 
+-export([queue_topology/1]).
+
 %% for backwards compatibility
 -export([file_handle_leader_reservation/1,
          file_handle_other_reservation/0,
@@ -2222,3 +2224,14 @@ maybe_log_leader_health_check_result([]) -> ok;
 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 :: undefined | node(), Replicas :: undefined | [node(),...]}.
+queue_topology(Q) ->
+    [{leader, Leader0},
+             {members, Members}] = rabbit_queue_type:info(Q, [leader, members]),
+            Leader = case Leader0 of
+                         '' -> undefined;
+                         _ -> Leader0
+                     end,
+            {Leader, Members}.

deps/rabbit/src/rabbit_stream_queue.erl

@@ -59,6 +59,8 @@
 
 -export([check_max_segment_size_bytes/1]).
 
+-export([queue_topology/1]).
+
 -include_lib("rabbit_common/include/rabbit.hrl").
 -include("amqqueue.hrl").
 
@@ -1393,3 +1395,18 @@ delivery_count_add(none, _) ->
     none;
 delivery_count_add(Count, N) ->
     serial_number:add(Count, N).
+
+-spec queue_topology(amqqueue:amqqueue()) ->
+    {Leader :: undefined | node(), Replicas :: undefined | [node(),...]}.
+queue_topology(Q) ->
+    #{name := StreamId} = amqqueue:get_type_state(Q),
+    case rabbit_stream_coordinator:members(StreamId) of
+        {ok, Members} ->
+            maps:fold(fun(Node, {_Pid, writer}, {_, Replicas}) ->
+                              {Node, [Node | Replicas]};
+                         (Node, {_Pid, replica}, {Writer, Replicas}) ->
+                              {Writer, [Node | Replicas]}
+                      end, {undefined, []}, Members);
+                {error, _} ->
+            {undefined, undefined}
+    end.

deps/rabbitmq_mqtt/src/rabbit_mqtt_qos0_queue.erl

@@ -41,6 +41,8 @@
          notify_decorators/1
         ]).
 
+-export([queue_topology/1]).
+
 %% Stateful rabbit_queue_type callbacks are unsupported by this queue type.
 -define(STATEFUL_CALLBACKS,
         [
@@ -301,3 +303,10 @@ dequeue(A1,A2,A3,A4,A5) ->
 
 state_info(A1) ->
     ?UNSUPPORTED([A1]).
+
+-spec queue_topology(amqqueue:amqqueue()) ->
+    {Leader :: undefined | node(), Replicas :: undefined | [node(),...]}.
+queue_topology(Q) ->
+    Pid = amqqueue:get_pid(Q),
+    Node = node(Pid),
+    {Node, [Node]}.