diff --git a/apps/emqx/src/emqx_session.erl b/apps/emqx/src/emqx_session.erl index 6567bf5c8..4d1594a99 100644 --- a/apps/emqx/src/emqx_session.erl +++ b/apps/emqx/src/emqx_session.erl @@ -47,6 +47,7 @@ -include("emqx_mqtt.hrl"). -include("logger.hrl"). -include("types.hrl"). +-include_lib("snabbkaffe/include/snabbkaffe.hrl"). -ifdef(TEST). -compile(export_all). @@ -802,7 +803,7 @@ replay(ClientInfo, Session = #session{inflight = Inflight}) -> -spec terminate(emqx_types:clientinfo(), Reason :: term(), session()) -> ok. terminate(ClientInfo, Reason, Session) -> run_terminate_hooks(ClientInfo, Reason, Session), - redispatch_shared_messages(Reason, Session), + maybe_redispatch_shared_messages(Reason, Session), ok. run_terminate_hooks(ClientInfo, discarded, Session) -> @@ -812,11 +813,13 @@ run_terminate_hooks(ClientInfo, takenover, Session) -> run_terminate_hooks(ClientInfo, Reason, Session) -> run_hook('session.terminated', [ClientInfo, Reason, info(Session)]). -redispatch_shared_messages(takenover, _Session) -> +maybe_redispatch_shared_messages(takenover, _Session) -> + ?tp(debug, ignore_redispatch_shared_messages, #{reason => takenover}), ok; -redispatch_shared_messages(kicked, _Session) -> +maybe_redispatch_shared_messages(kicked, _Session) -> + ?tp(debug, ignore_redispatch_shared_messages, #{reason => kicked}), ok; -redispatch_shared_messages(_Reason, #session{inflight = Inflight, mqueue = Q}) -> +maybe_redispatch_shared_messages(_Reason, #session{inflight = Inflight, mqueue = Q}) -> AllInflights = emqx_inflight:to_list(fun sort_fun/2, Inflight), F = fun ({_PacketId, #inflight_data{message = #message{qos = ?QOS_1} = Msg}}) -> diff --git a/apps/emqx/test/emqx_shared_sub_SUITE.erl b/apps/emqx/test/emqx_shared_sub_SUITE.erl index 2cdf2bc05..f048df397 100644 --- a/apps/emqx/test/emqx_shared_sub_SUITE.erl +++ b/apps/emqx/test/emqx_shared_sub_SUITE.erl @@ -22,6 +22,7 @@ -include_lib("emqx/include/emqx.hrl"). -include_lib("eunit/include/eunit.hrl"). -include_lib("common_test/include/ct.hrl"). +-include_lib("snabbkaffe/include/snabbkaffe.hrl"). -define(SUITE, ?MODULE). @@ -869,23 +870,23 @@ t_redispatch_when_kicked(_) -> Subs = [erlang:spawn(fun() -> MkSub(ClientId) end) || ClientId <- [ClientId1, ClientId2]], timer:sleep(500), + ok = snabbkaffe:start_trace(), + Message = emqx_message:make(ClientId1, 1, Topic, <<"hello1">>), emqx:publish(Message), [Sub ! wait || Sub <- Subs], - timer:sleep(200), Res = Receive(), ?assertMatch({got, _}, Res), {got, ClientId} = Res, emqx_cm:kick_session(ClientId), - [Sub ! wait || Sub <- Subs], - timer:sleep(200), - Res1 = Receive(), - ?assertMatch(nothing, Res1), + Trace = snabbkaffe:collect_trace(500), + ?assertMatch([#{reason := kicked}], ?of_kind(ignore_redispatch_shared_messages, Trace)), [Sub ! stop || Sub <- Subs], + snabbkaffe:stop(), ok. %%--------------------------------------------------------------------