Skip to content

Commit 9bb622c

Browse files
ansdansd
committed
Add queue topology test
1 parent 6de51ee commit 9bb622c

File tree

2 files changed

+98
-22
lines changed

2 files changed

+98
-22
lines changed

‎deps/rabbit/src/rabbit_amqp_management.erl‎

Lines changed: 30 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -372,7 +372,6 @@ encode_queue(Q, NumMsgs, NumConsumers) ->
372372
QArgs = args_amqpl_to_amqp(QArgs091),
373373
{Leader, Replicas} = queue_topology(Q),
374374
KVList0 = [
375-
{{utf8, <<"replicas">>}, {array, utf8, [{utf8, atom_to_binary(R)} || R <- Replicas]}},
376375
{{utf8, <<"message_count">>}, {ulong, NumMsgs}},
377376
{{utf8, <<"consumer_count">>}, {uint, NumConsumers}},
378377
{{utf8, <<"name">>}, {utf8, QNameBin}},
@@ -383,28 +382,47 @@ encode_queue(Q, NumMsgs, NumConsumers) ->
383382
{{utf8, <<"type">>}, {utf8, rabbit_queue_type:to_binary(QType)}},
384383
{{utf8, <<"arguments">>}, QArgs}
385384
],
385+
KVList1 = if is_list(Replicas) ->
386+
[{{utf8, <<"replicas">>},
387+
{array, utf8, [{utf8, atom_to_binary(R)} || R <- Replicas]}
388+
} | KVList0];
389+
Replicas =:= undefined ->
390+
KVList0
391+
end,
386392
KVList = if is_atom(Leader) ->
387-
[{{utf8, <<"leader">>}, {utf8, atom_to_binary(Leader)}} | KVList0];
393+
[{{utf8, <<"leader">>},
394+
{utf8, atom_to_binary(Leader)}
395+
} | KVList1];
388396
Leader =:= undefined ->
389-
KVList0
397+
KVList1
390398
end,
391399
{map, KVList}.
392400

401+
%% The returned Replicas contain both online and offline replicas.
393402
-spec queue_topology(amqqueue:amqqueue()) ->
394-
{Leader :: undefined | node(), Replicas :: [node(),...]}.
403+
{Leader :: undefined | node(), Replicas :: undefined | [node(),...]}.
395404
queue_topology(Q) ->
396405
case amqqueue:get_type(Q) of
397406
rabbit_quorum_queue ->
398-
Leader = amqqueue:get_leader(Q),
399-
[{members, Members}] = rabbit_quorum_queue:info(Q, [members]),
407+
[{leader, Leader0},
408+
{members, Members}] = rabbit_queue_type:info(Q, [leader, members]),
409+
Leader = case Leader0 of
410+
'' -> undefined;
411+
_ -> Leader0
412+
end,
400413
{Leader, Members};
401414
rabbit_stream_queue ->
402-
%% TODO How to best determine writer and replicas?
403-
% #{name := StreamId} = amqqueue:get_type_state(Q),
404-
% rabbit_stream_coordinator:members(StreamId)
405-
[{leader, Leader},
406-
{members, Members}] = rabbit_stream_queue:info(Q, [leader, members]),
407-
{Leader, Members};
415+
#{name := StreamId} = amqqueue:get_type_state(Q),
416+
case rabbit_stream_coordinator:members(StreamId) of
417+
{ok, Members} ->
418+
maps:fold(fun(Node, {_Pid, writer}, {_, Replicas}) ->
419+
{Node, [Node | Replicas]};
420+
(Node, {_Pid, replica}, {Writer, Replicas}) ->
421+
{Writer, [Node | Replicas]}
422+
end, {undefined, []}, Members);
423+
{error, _} ->
424+
{undefined, undefined}
425+
end;
408426
_ ->
409427
Pid = amqqueue:get_pid(Q),
410428
Node = node(Pid),

‎deps/rabbitmq_amqp_client/test/management_SUITE.erl‎

Lines changed: 68 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -75,7 +75,8 @@ groups() ->
7575
purge_stream
7676
]},
7777
{cluster_size_3, [shuffle],
78-
[classic_queue_stopped
78+
[classic_queue_stopped,
79+
queue_topology
7980
]}
8081
].
8182

@@ -563,19 +564,15 @@ declare_exchange_inequivalent_fields(Config) ->
563564
ok = cleanup(Init).
564565

565566
classic_queue_stopped(Config) ->
566-
OpnConf2 = connection_config(Config, 2),
567-
{ok, Conn2} = amqp10_client:open_connection(OpnConf2),
568-
{ok, Session2} = amqp10_client:begin_session_sync(Conn2),
569-
{ok, LinkPair2} = rabbitmq_amqp_client:attach_management_link_pair_sync(Session2, <<"my link pair">>),
570-
567+
Init2 = {_, _, LinkPair2} = init(Config, 2),
571568
QName = <<"👌"/utf8>>,
572569
{ok, #{durable := true,
573570
type := <<"classic">>}} = rabbitmq_amqp_client:declare_queue(LinkPair2, QName, #{}),
574-
ok = amqp10_client:close_connection(Conn2),
571+
ok = cleanup(Init2),
575572
ok = rabbit_ct_broker_helpers:stop_node(Config, 2),
576573
%% Classic queue is now stopped.
577574

578-
Init = {_, _, LinkPair0} = init(Config),
575+
Init0 = {_, _, LinkPair0} = init(Config),
579576
{error, Resp0} = rabbitmq_amqp_client:declare_queue(LinkPair0, QName, #{}),
580577
?assertMatch(#{subject := <<"400">>}, amqp10_msg:properties(Resp0)),
581578
ExpectedResponseBody = #'v1_0.amqp_value'{
@@ -591,7 +588,7 @@ classic_queue_stopped(Config) ->
591588

592589
ok = rabbit_ct_broker_helpers:start_node(Config, 2),
593590
{ok, #{}} = rabbitmq_amqp_client:delete_queue(LinkPair0, QName),
594-
ok = cleanup(Init).
591+
ok = cleanup(Init0).
595592

596593
delete_default_exchange(Config) ->
597594
Init = {_, _, LinkPair} = init(Config),
@@ -731,8 +728,69 @@ purge_stream(Config) ->
731728
{ok, #{}} = rabbitmq_amqp_client:delete_queue(LinkPair, QName),
732729
ok = cleanup(Init).
733730

731+
queue_topology(Config) ->
732+
NodeNames = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
733+
Nodes = [N0, N1, N2] = lists:map(fun erlang:atom_to_binary/1, NodeNames),
734+
Init0 = {_, _, LinkPair0} = init(Config, 0),
735+
736+
CQName = <<"my classic queue">>,
737+
QQName = <<"my quorum queue">>,
738+
SQName = <<"my stream queue">>,
739+
740+
CQProps = #{arguments => #{<<"x-queue-type">> => {utf8, <<"classic">>}}},
741+
QQProps = #{arguments => #{<<"x-queue-type">> => {utf8, <<"quorum">>}}},
742+
SQProps = #{arguments => #{<<"x-queue-type">> => {utf8, <<"stream">>}}},
743+
744+
{ok, CQInfo0} = rabbitmq_amqp_client:declare_queue(LinkPair0, CQName, CQProps),
745+
{ok, QQInfo0} = rabbitmq_amqp_client:declare_queue(LinkPair0, QQName, QQProps),
746+
{ok, SQInfo0} = rabbitmq_amqp_client:declare_queue(LinkPair0, SQName, SQProps),
747+
748+
%% The default queue leader strategy is client-local.
749+
?assertEqual({ok, N0}, maps:find(leader, CQInfo0)),
750+
?assertEqual({ok, N0}, maps:find(leader, QQInfo0)),
751+
?assertEqual({ok, N0}, maps:find(leader, SQInfo0)),
752+
753+
?assertEqual({ok, [N0]}, maps:find(replicas, CQInfo0)),
754+
{ok, QQReplicas0} = maps:find(replicas, QQInfo0),
755+
?assertEqual(Nodes, lists:usort(QQReplicas0)),
756+
{ok, SQReplicas0} = maps:find(replicas, SQInfo0),
757+
?assertEqual(Nodes, lists:usort(SQReplicas0)),
758+
759+
ok = cleanup(Init0),
760+
ok = rabbit_ct_broker_helpers:stop_node(Config, 0),
761+
762+
Init2 = {_, _, LinkPair2} = init(Config, 2),
763+
{ok, QQInfo2} = rabbitmq_amqp_client:get_queue(LinkPair2, QQName),
764+
{ok, SQInfo2} = rabbitmq_amqp_client:get_queue(LinkPair2, SQName),
765+
766+
case maps:get(leader, QQInfo2) of
767+
N1 -> ok;
768+
N2 -> ok;
769+
Other0 -> ct:fail({?LINE, Other0})
770+
end,
771+
case maps:get(leader, SQInfo2) of
772+
N1 -> ok;
773+
N2 -> ok;
774+
Other1 -> ct:fail({?LINE, Other1})
775+
end,
776+
777+
%% Replicas should include both online and offline replicas.
778+
{ok, QQReplicas2} = maps:find(replicas, QQInfo2),
779+
?assertEqual(Nodes, lists:usort(QQReplicas2)),
780+
{ok, SQReplicas2} = maps:find(replicas, SQInfo2),
781+
?assertEqual(Nodes, lists:usort(SQReplicas2)),
782+
783+
ok = rabbit_ct_broker_helpers:start_node(Config, 0),
784+
{ok, _} = rabbitmq_amqp_client:delete_queue(LinkPair2, CQName),
785+
{ok, _} = rabbitmq_amqp_client:delete_queue(LinkPair2, QQName),
786+
{ok, _} = rabbitmq_amqp_client:delete_queue(LinkPair2, SQName),
787+
ok = cleanup(Init2).
788+
734789
init(Config) ->
735-
OpnConf = connection_config(Config),
790+
init(Config, 0).
791+
792+
init(Config, Node) ->
793+
OpnConf = connection_config(Config, Node),
736794
{ok, Connection} = amqp10_client:open_connection(OpnConf),
737795
{ok, Session} = amqp10_client:begin_session_sync(Connection),
738796
{ok, LinkPair} = rabbitmq_amqp_client:attach_management_link_pair_sync(Session, <<"my link pair">>),

0 commit comments

Comments
 (0)