Dirty update testcases for shared_sub
This commit is contained in:
terry-xiaoyu
parent
857b3df93d
commit
2fcda7d891
|
|
@ -24,6 +24,7 @@
|
|||
-include_lib("common_test/include/ct.hrl").
|
||||
|
||||
-define(SUITE, ?MODULE).
|
||||
|
||||
-define(wait(For, Timeout),
|
||||
emqx_ct_helpers:wait_for(
|
||||
?FUNCTION_NAME, ?LINE, fun() -> For end, Timeout)).
|
||||
|
|
@ -40,21 +41,23 @@ end_per_suite(_Config) ->
|
|||
t_random_basic(_) ->
|
||||
ok = ensure_config(random),
|
||||
ClientId = <<"ClientId">>,
|
||||
{ok, ConnPid} = emqx_mock_client:start_link(ClientId),
|
||||
{ok, SPid} = emqx_mock_client:open_session(ConnPid, ClientId, internal),
|
||||
Message1 = emqx_message:make(<<"ClientId">>, 2, <<"foo">>, <<"hello">>),
|
||||
emqx_session:subscribe(SPid, [{<<"foo">>, #{qos => 2, share => <<"group1">>}}]),
|
||||
Topic = <<"foo">>,
|
||||
Payload = <<"hello">>,
|
||||
emqx:subscribe(Topic, #{qos => 2, share => <<"group1">>}),
|
||||
MsgQoS2 = emqx_message:make(ClientId, 2, Topic, Payload),
|
||||
%% wait for the subscription to show up
|
||||
?wait(subscribed(<<"group1">>, <<"foo">>, SPid), 1000),
|
||||
PacketId = 1,
|
||||
emqx_session:publish(SPid, PacketId, Message1),
|
||||
?wait(case emqx_mock_client:get_last_message(ConnPid) of
|
||||
[{publish, 1, _}] -> true;
|
||||
Other -> Other
|
||||
end, 1000),
|
||||
emqx_session:pubrec(SPid, PacketId, reasoncode),
|
||||
emqx_session:pubcomp(SPid, PacketId, reasoncode),
|
||||
emqx_mock_client:close_session(ConnPid),
|
||||
ct:sleep(200),
|
||||
?assertEqual(true, subscribed(<<"group1">>, Topic, self())),
|
||||
emqx:publish(MsgQoS2),
|
||||
receive
|
||||
{deliver, Topic0, #message{from = ClientId0,
|
||||
payload = Payload0}} = M->
|
||||
ct:pal("==== received: ~p", [M]),
|
||||
?assertEqual(Topic, Topic0),
|
||||
?assertEqual(ClientId, ClientId0),
|
||||
?assertEqual(Payload, Payload0)
|
||||
after 1000 -> ct:fail(waiting_basic_failed)
|
||||
end,
|
||||
ok.
|
||||
|
||||
%% Start two subscribers share subscribe to "$share/g1/foo/bar"
|
||||
|
|
@ -73,69 +76,61 @@ t_no_connection_nack(_) ->
|
|||
QoS = 1,
|
||||
Group = <<"g1">>,
|
||||
Topic = <<"foo/bar">>,
|
||||
{ok, PubConnPid} = emqx_mock_client:start_link(Publisher),
|
||||
{ok, SubConnPid1} = emqx_mock_client:start_link(Subscriber1),
|
||||
{ok, SubConnPid2} = emqx_mock_client:start_link(Subscriber2),
|
||||
%% allow session to persist after connection shutdown
|
||||
Attrs = #{expiry_interval => timer:seconds(30)},
|
||||
{ok, P_Pid} = emqx_mock_client:open_session(PubConnPid, Publisher, internal, Attrs),
|
||||
{ok, SPid1} = emqx_mock_client:open_session(SubConnPid1, Subscriber1, internal, Attrs),
|
||||
{ok, SPid2} = emqx_mock_client:open_session(SubConnPid2, Subscriber2, internal, Attrs),
|
||||
emqx_session:subscribe(SPid1, [{Topic, #{qos => QoS, share => Group}}]),
|
||||
emqx_session:subscribe(SPid2, [{Topic, #{qos => QoS, share => Group}}]),
|
||||
ShareTopic = <<"$share/", Group/binary, $/, Topic/binary>>,
|
||||
|
||||
ExpProp = [{properties, #{'Session-Expiry-Interval' => timer:seconds(30)}}],
|
||||
{ok, SubConnPid1} = emqtt:start_link([{client_id, Subscriber1}] ++ ExpProp),
|
||||
{ok, _Props} = emqtt:connect(SubConnPid1),
|
||||
{ok, SubConnPid2} = emqtt:start_link([{client_id, Subscriber2}] ++ ExpProp),
|
||||
{ok, _Props} = emqtt:connect(SubConnPid2),
|
||||
emqtt:subscribe(SubConnPid1, ShareTopic, QoS),
|
||||
emqtt:subscribe(SubConnPid1, ShareTopic, QoS),
|
||||
|
||||
%% wait for the subscriptions to show up
|
||||
?wait(subscribed(Group, Topic, SPid1), 1000),
|
||||
?wait(subscribed(Group, Topic, SPid2), 1000),
|
||||
MkPayload = fun(PacketId) -> iolist_to_binary(["hello-", integer_to_list(PacketId)]) end,
|
||||
SendF = fun(PacketId) -> emqx_session:publish(P_Pid, PacketId, emqx_message:make(Publisher, QoS, Topic, MkPayload(PacketId))) end,
|
||||
ct:sleep(200),
|
||||
MkPayload = fun(PacketId) ->
|
||||
iolist_to_binary(["hello-", integer_to_list(PacketId)])
|
||||
end,
|
||||
SendF = fun(PacketId) ->
|
||||
M = emqx_message:make(Publisher, QoS, Topic, MkPayload(PacketId)),
|
||||
emqx:publish(M#message{id = PacketId})
|
||||
end,
|
||||
SendF(1),
|
||||
Ref = make_ref(),
|
||||
CasePid = self(),
|
||||
Received =
|
||||
fun(PacketId, ConnPid) ->
|
||||
Payload = MkPayload(PacketId),
|
||||
case emqx_mock_client:get_last_message(ConnPid) of
|
||||
[{publish, _, #message{payload = Payload}}] ->
|
||||
CasePid ! {Ref, PacketId, ConnPid},
|
||||
true;
|
||||
_Other ->
|
||||
false
|
||||
end
|
||||
end,
|
||||
?wait(Received(1, SubConnPid1) orelse Received(1, SubConnPid2), 1000),
|
||||
timer:sleep(200),
|
||||
%% This is the connection which was picked by broker to dispatch (sticky) for 1st message
|
||||
ConnPid = receive {Ref, 1, Pid} -> Pid after 1000 -> error(timeout) end,
|
||||
|
||||
?assertMatch([#{packet_id := 1}], recv_msgs(1)),
|
||||
%% Now kill the connection, expect all following messages to be delivered to the other subscriber.
|
||||
emqx_mock_client:stop(ConnPid),
|
||||
%emqx_mock_client:stop(ConnPid),
|
||||
%% sleep then make synced calls to session processes to ensure that
|
||||
%% the connection pid's 'EXIT' message is propagated to the session process
|
||||
%% also to be sure sessions are still alive
|
||||
timer:sleep(2),
|
||||
_ = emqx_session:info(SPid1),
|
||||
_ = emqx_session:info(SPid2),
|
||||
%% Now we know what is the other still alive connection
|
||||
[TheOtherConnPid] = [SubConnPid1, SubConnPid2] -- [ConnPid],
|
||||
%% Send some more messages
|
||||
PacketIdList = lists:seq(2, 10),
|
||||
lists:foreach(fun(Id) ->
|
||||
SendF(Id),
|
||||
?wait(Received(Id, TheOtherConnPid), 1000)
|
||||
end, PacketIdList),
|
||||
%% Now close the 2nd (last connection)
|
||||
emqx_mock_client:stop(TheOtherConnPid),
|
||||
timer:sleep(2),
|
||||
%% both sessions should have conn_pid = undefined
|
||||
?assertEqual({conn_pid, undefined}, lists:keyfind(conn_pid, 1, emqx_session:info(SPid1))),
|
||||
?assertEqual({conn_pid, undefined}, lists:keyfind(conn_pid, 1, emqx_session:info(SPid2))),
|
||||
%% send more messages, but all should be queued in session state
|
||||
lists:foreach(fun(Id) -> SendF(Id) end, PacketIdList),
|
||||
{_, L1} = lists:keyfind(mqueue_len, 1, emqx_session:info(SPid1)),
|
||||
{_, L2} = lists:keyfind(mqueue_len, 1, emqx_session:info(SPid2)),
|
||||
?assertEqual(length(PacketIdList), L1 + L2),
|
||||
%% clean up
|
||||
emqx_mock_client:close_session(PubConnPid),
|
||||
emqx_sm:close_session(SPid1),
|
||||
emqx_sm:close_session(SPid2),
|
||||
% timer:sleep(2),
|
||||
% _ = emqx_session:info(SPid1),
|
||||
% _ = emqx_session:info(SPid2),
|
||||
% %% Now we know what is the other still alive connection
|
||||
% [TheOtherConnPid] = [SubConnPid1, SubConnPid2] -- [ConnPid],
|
||||
% %% Send some more messages
|
||||
% PacketIdList = lists:seq(2, 10),
|
||||
% lists:foreach(fun(Id) ->
|
||||
% SendF(Id),
|
||||
% ?wait(Received(Id, TheOtherConnPid), 1000)
|
||||
% end, PacketIdList),
|
||||
% %% Now close the 2nd (last connection)
|
||||
% emqx_mock_client:stop(TheOtherConnPid),
|
||||
% timer:sleep(2),
|
||||
% %% both sessions should have conn_pid = undefined
|
||||
% ?assertEqual({conn_pid, undefined}, lists:keyfind(conn_pid, 1, emqx_session:info(SPid1))),
|
||||
% ?assertEqual({conn_pid, undefined}, lists:keyfind(conn_pid, 1, emqx_session:info(SPid2))),
|
||||
% %% send more messages, but all should be queued in session state
|
||||
% lists:foreach(fun(Id) -> SendF(Id) end, PacketIdList),
|
||||
% {_, L1} = lists:keyfind(mqueue_len, 1, emqx_session:info(SPid1)),
|
||||
% {_, L2} = lists:keyfind(mqueue_len, 1, emqx_session:info(SPid2)),
|
||||
% ?assertEqual(length(PacketIdList), L1 + L2),
|
||||
% %% clean up
|
||||
% emqx_mock_client:close_session(PubConnPid),
|
||||
% emqx_sm:close_session(SPid1),
|
||||
% emqx_sm:close_session(SPid2),
|
||||
ok.
|
||||
|
||||
t_random(_) ->
|
||||
|
|
@ -155,31 +150,24 @@ t_not_so_sticky(_) ->
|
|||
ok = ensure_config(sticky),
|
||||
ClientId1 = <<"ClientId1">>,
|
||||
ClientId2 = <<"ClientId2">>,
|
||||
{ok, ConnPid1} = emqx_mock_client:start_link(ClientId1),
|
||||
{ok, ConnPid2} = emqx_mock_client:start_link(ClientId2),
|
||||
{ok, SPid1} = emqx_mock_client:open_session(ConnPid1, ClientId1, internal),
|
||||
{ok, SPid2} = emqx_mock_client:open_session(ConnPid2, ClientId2, internal),
|
||||
Message1 = emqx_message:make(ClientId1, 0, <<"foo/bar">>, <<"hello1">>),
|
||||
Message2 = emqx_message:make(ClientId1, 0, <<"foo/bar">>, <<"hello2">>),
|
||||
emqx_session:subscribe(SPid1, [{<<"foo/bar">>, #{qos => 0, share => <<"group1">>}}]),
|
||||
%% wait for the subscription to show up
|
||||
?wait(subscribed(<<"group1">>, <<"foo/bar">>, SPid1), 1000),
|
||||
emqx_session:publish(SPid1, 1, Message1),
|
||||
?wait(case emqx_mock_client:get_last_message(ConnPid1) of
|
||||
[{publish, _, #message{payload = <<"hello1">>}}] -> true;
|
||||
Other -> Other
|
||||
end, 1000),
|
||||
emqx_mock_client:close_session(ConnPid1),
|
||||
?wait(not subscribed(<<"group1">>, <<"foo/bar">>, SPid1), 1000),
|
||||
emqx_session:subscribe(SPid2, [{<<"foo/#">>, #{qos => 0, share => <<"group1">>}}]),
|
||||
?wait(subscribed(<<"group1">>, <<"foo/#">>, SPid2), 1000),
|
||||
emqx_session:publish(SPid2, 2, Message2),
|
||||
?wait(case emqx_mock_client:get_last_message(ConnPid2) of
|
||||
[{publish, _, #message{payload = <<"hello2">>}}] -> true;
|
||||
Other -> Other
|
||||
end, 1000),
|
||||
emqx_mock_client:close_session(ConnPid2),
|
||||
?wait(not subscribed(<<"group1">>, <<"foo/#">>, SPid2), 1000),
|
||||
{ok, C1} = emqx_client:start_link([{client_id, ClientId1}]),
|
||||
{ok, _} = emqx_client:connect(C1),
|
||||
{ok, C2} = emqx_client:start_link([{client_id, ClientId2}]),
|
||||
{ok, _} = emqx_client:connect(C2),
|
||||
|
||||
emqx_client:subscribe(C1, {<<"$share/group1/foo/bar">>, 0}),
|
||||
timer:sleep(50),
|
||||
emqx_client:publish(C2, <<"foo/bar">>, <<"hello1">>),
|
||||
?assertMatch([#{payload := <<"hello1">>}], recv_msgs(1)),
|
||||
|
||||
emqx_client:unsubscribe(C1, <<"$share/group1/foo/bar">>),
|
||||
timer:sleep(50),
|
||||
emqx_client:subscribe(C1, {<<"$share/group1/foo/#">>, 0}),
|
||||
timer:sleep(50),
|
||||
emqx_client:publish(C2, <<"foo/bar">>, <<"hello2">>),
|
||||
?assertMatch([#{payload := <<"hello2">>}], recv_msgs(1)),
|
||||
emqx_client:disconnect(C1),
|
||||
emqx_client:disconnect(C2),
|
||||
ok.
|
||||
|
||||
test_two_messages(Strategy) ->
|
||||
|
|
@ -190,18 +178,17 @@ test_two_messages(Strategy, WithAck) ->
|
|||
Topic = <<"foo/bar">>,
|
||||
ClientId1 = <<"ClientId1">>,
|
||||
ClientId2 = <<"ClientId2">>,
|
||||
{ok, ConnPid1} = emqx_mock_client:start_link(ClientId1),
|
||||
{ok, ConnPid2} = emqx_mock_client:start_link(ClientId2),
|
||||
{ok, SPid1} = emqx_mock_client:open_session(ConnPid1, ClientId1, internal),
|
||||
{ok, SPid2} = emqx_mock_client:open_session(ConnPid2, ClientId2, internal),
|
||||
{ok, ConnPid1} = emqx_client:start_link([{client_id, ClientId1}]),
|
||||
{ok, _} = emqx_client:connect(ConnPid1),
|
||||
{ok, ConnPid2} = emqx_client:start_link([{client_id, ClientId2}]),
|
||||
{ok, _} = emqx_client:connect(ConnPid2),
|
||||
|
||||
Message1 = emqx_message:make(ClientId1, 0, Topic, <<"hello1">>),
|
||||
Message2 = emqx_message:make(ClientId1, 0, Topic, <<"hello2">>),
|
||||
emqx_session:subscribe(SPid1, [{Topic, #{qos => 0, share => <<"group1">>}}]),
|
||||
emqx_session:subscribe(SPid2, [{Topic, #{qos => 0, share => <<"group1">>}}]),
|
||||
%% wait for the subscription to show up
|
||||
?wait(subscribed(<<"group1">>, Topic, SPid1) andalso
|
||||
subscribed(<<"group1">>, Topic, SPid2), 1000),
|
||||
emqx_broker:publish(Message1),
|
||||
emqx_client:subscribe(ConnPid1, {<<"$share/group1/foo/bar">>, 0}),
|
||||
emqx_client:subscribe(ConnPid2, {<<"$share/group1/foo/bar">>, 0}),
|
||||
ct:sleep(100),
|
||||
emqx:publish(Message1),
|
||||
Me = self(),
|
||||
WaitF = fun(ExpectedPayload) ->
|
||||
case last_message(ExpectedPayload, [ConnPid1, ConnPid2]) of
|
||||
|
|
@ -212,10 +199,10 @@ test_two_messages(Strategy, WithAck) ->
|
|||
Other
|
||||
end
|
||||
end,
|
||||
?wait(WaitF(<<"hello1">>), 2000),
|
||||
WaitF(<<"hello1">>),
|
||||
UsedSubPid1 = receive {subscriber, P1} -> P1 end,
|
||||
emqx_broker:publish(Message2),
|
||||
?wait(WaitF(<<"hello2">>), 2000),
|
||||
WaitF(<<"hello2">>),
|
||||
UsedSubPid2 = receive {subscriber, P2} -> P2 end,
|
||||
case Strategy of
|
||||
sticky -> ?assert(UsedSubPid1 =:= UsedSubPid2);
|
||||
|
|
@ -223,15 +210,17 @@ test_two_messages(Strategy, WithAck) ->
|
|||
hash -> ?assert(UsedSubPid1 =:= UsedSubPid2);
|
||||
_ -> ok
|
||||
end,
|
||||
emqx_mock_client:close_session(ConnPid1),
|
||||
emqx_mock_client:close_session(ConnPid2),
|
||||
emqx_client:stop(ConnPid1),
|
||||
emqx_client:stop(ConnPid2),
|
||||
ok.
|
||||
|
||||
last_message(_ExpectedPayload, []) -> <<"not yet?">>;
|
||||
last_message(ExpectedPayload, [Pid | Pids]) ->
|
||||
case emqx_mock_client:get_last_message(Pid) of
|
||||
[{publish, _, #message{payload = ExpectedPayload}}] -> {true, Pid};
|
||||
_Other -> last_message(ExpectedPayload, Pids)
|
||||
last_message(ExpectedPayload, Pids) ->
|
||||
receive
|
||||
{publish, #{client_pid := Pid, payload := ExpectedPayload}} ->
|
||||
ct:pal("~p ====== ~p", [Pids, Pid]),
|
||||
{true, Pid}
|
||||
after 100 ->
|
||||
<<"not yet?">>
|
||||
end.
|
||||
|
||||
%%--------------------------------------------------------------------
|
||||
|
|
@ -249,3 +238,16 @@ ensure_config(Strategy, AckEnabled) ->
|
|||
subscribed(Group, Topic, Pid) ->
|
||||
lists:member(Pid, emqx_shared_sub:subscribers(Group, Topic)).
|
||||
|
||||
recv_msgs(Count) ->
|
||||
recv_msgs(Count, []).
|
||||
|
||||
recv_msgs(0, Msgs) ->
|
||||
Msgs;
|
||||
recv_msgs(Count, Msgs) ->
|
||||
receive
|
||||
{publish, Msg} ->
|
||||
recv_msgs(Count-1, [Msg|Msgs]);
|
||||
_Other -> recv_msgs(Count, Msgs) %%TODO:: remove the branch?
|
||||
after 100 ->
|
||||
Msgs
|
||||
end.
|
||||
|
|
|
|||
Loading…
Reference in New Issue