Queues with plugins - sync with queue topologies updates

deps/rabbit/src/rabbit_classic_queue.erl

@@ -702,7 +702,7 @@ 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(),...]}.
+    {Leader :: node() | none, Replicas :: [node(),...]}.
 queue_topology(Q) ->
     Pid = amqqueue:get_pid(Q),
     Node = node(Pid),

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -2253,7 +2253,7 @@ maybe_log_leader_health_check_result(Result) ->
     rabbit_log:warning("Leader health check result (unhealthy leaders detected): ~tp", [Qs]).
 
 -spec queue_topology(amqqueue:amqqueue()) ->
-    {Leader :: undefined | node(), Replicas :: undefined | [node(),...]}.
+          {Leader :: node() | none, Replicas :: [node(),...]}.
 queue_topology(Q) ->
     Leader = case amqqueue:get_pid(Q) of
                  {_RaName, Node} ->

deps/rabbit/src/rabbit_stream_queue.erl

@@ -1430,19 +1430,23 @@ delivery_count_add(Count, N) ->
     serial_number:add(Count, N).
 
 -spec queue_topology(amqqueue:amqqueue()) ->
-    {Leader :: undefined | node(), Replicas :: undefined | [node(),...]}.
+    {Leader :: node() | none, Replicas :: [node(),...]}.
 queue_topology(Q) ->
-    #{name := StreamId} = amqqueue:get_type_state(Q),
+    Leader = case amqqueue:get_pid(Q) of
-    case rabbit_stream_coordinator:members(StreamId) of
+                 {_RaName, Node} ->
-        {ok, Members} ->
+                     Node;
-            maps:fold(fun(Node, {_Pid, writer}, {_, Replicas}) ->
+                 none ->
-                              {Node, [Node | Replicas]};
+                     none;
-                         (Node, {_Pid, replica}, {Writer, Replicas}) ->
+                 Pid ->
-                              {Writer, [Node | Replicas]}
+                     node(Pid)
-                      end, {undefined, []}, Members);
+             end,
-                {error, _} ->
+    Replicas = case amqqueue:get_type_state(Q) of
-            {undefined, undefined}
+                   #{nodes := Nodes} ->
-    end.
+                       Nodes;
+                   _ ->
+                       [Leader]
+               end,
+    {Leader, Replicas}.
 
 policy_apply_to_name() ->
     <<"streams">>.