yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

test(persistent_session): try to fix flaky snabbkaffe failure

This commit is contained in:
Tobias Lindahl 2021-11-01 10:17:00 +01:00
parent b4d99bc176
commit bb040ed95e
1 changed files with 10 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
apps/emqx/test/emqx_persistent_session_SUITE.erl
View File

@ -265,33 +265,37 @@ snabbkaffe_sync_publish(Topic, Payloads) ->
, #{?snk_kind := ps_persist_msg, payload := Payload} , #{?snk_kind := ps_persist_msg, payload := Payload}
) )
end, end,
do_publish(Payloads, Fun). do_publish(Payloads, Fun, true).
publish(Topic, Payloads) -> publish(Topic, Payloads) ->
Fun = fun(Client, Payload) -> Fun = fun(Client, Payload) ->
{ok, _} = emqtt:publish(Client, Topic, Payload, 2) {ok, _} = emqtt:publish(Client, Topic, Payload, 2)
end, end,
do_publish(Payloads, Fun). do_publish(Payloads, Fun, false).
do_publish(Payloads = [_|_], PublishFun) -> do_publish(Payloads = [_|_], PublishFun, WaitForUnregister) ->
%% Publish from another process to avoid connection confusion. %% Publish from another process to avoid connection confusion.
{Pid, Ref} = {Pid, Ref} =
spawn_monitor( spawn_monitor(
fun() -> fun() ->
%% For convenience, always publish using tcp. %% For convenience, always publish using tcp.
%% The publish path is not what we are testing. %% The publish path is not what we are testing.
ClientID = <<"ps_SUITE_publisher">>,
{ok, Client} = emqtt:start_link([ {proto_ver, v5} {ok, Client} = emqtt:start_link([ {proto_ver, v5}
, {clientid, ClientID}
, {port, 1883} ]), , {port, 1883} ]),
{ok, _} = emqtt:connect(Client), {ok, _} = emqtt:connect(Client),
lists:foreach(fun(Payload) -> PublishFun(Client, Payload) end, Payloads), lists:foreach(fun(Payload) -> PublishFun(Client, Payload) end, Payloads),
ok = emqtt:disconnect(Client) ok = emqtt:disconnect(Client),
%% Snabbkaffe sometimes fails unless all processes are gone.
[wait_for_cm_unregister(ClientID) || WaitForUnregister]
end), end),
receive receive
{'DOWN', Ref, process, Pid, normal} -> ok; {'DOWN', Ref, process, Pid, normal} -> ok;
{'DOWN', Ref, process, Pid, What} -> error({failed_publish, What}) {'DOWN', Ref, process, Pid, What} -> error({failed_publish, What})
end; end;
do_publish(Payload, PublishFun) -> do_publish(Payload, PublishFun, WaitForUnregister) ->
do_publish([Payload], PublishFun). do_publish([Payload], PublishFun, WaitForUnregister).
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% Test Cases %% Test Cases