diff --git a/apps/emqx/src/emqx_session.erl b/apps/emqx/src/emqx_session.erl index b285d0a88..6567bf5c8 100644 --- a/apps/emqx/src/emqx_session.erl +++ b/apps/emqx/src/emqx_session.erl @@ -802,8 +802,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), - Reason =/= takenover andalso - redispatch_shared_messages(Session), + redispatch_shared_messages(Reason, Session), ok. run_terminate_hooks(ClientInfo, discarded, Session) -> @@ -813,7 +812,11 @@ run_terminate_hooks(ClientInfo, takenover, Session) -> run_terminate_hooks(ClientInfo, Reason, Session) -> run_hook('session.terminated', [ClientInfo, Reason, info(Session)]). -redispatch_shared_messages(#session{inflight = Inflight, mqueue = Q}) -> +redispatch_shared_messages(takenover, _Session) -> + ok; +redispatch_shared_messages(kicked, _Session) -> + ok; +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 291286aa2..2cdf2bc05 100644 --- a/apps/emqx/test/emqx_shared_sub_SUITE.erl +++ b/apps/emqx/test/emqx_shared_sub_SUITE.erl @@ -817,6 +817,77 @@ t_dispatch_qos0(Config) when is_list(Config) -> emqtt:stop(ConnPid2), ok. +t_redispatch_when_kicked({init, Config}) when is_list(Config) -> + Config; +t_redispatch_when_kicked({'end', Config}) when is_list(Config) -> + ok; +t_redispatch_when_kicked(_) -> + ok = ensure_config(sticky, true), + + Group = <<"group1">>, + Topic = <<"foo/bar">>, + ClientId1 = <<"ClientId1">>, + ClientId2 = <<"ClientId2">>, + Parent = self(), + + %% emqx_cm:kick_session will cause the emqtt to exit abnormally + %% so here need to isolate emqtt with a separate process + MkSub = fun(ClientId) -> + {ok, ConnPid} = emqtt:start_link([{clientid, ClientId}, {auto_ack, false}]), + {ok, _} = emqtt:connect(ConnPid), + {ok, _Props, _ReasonCodes} = emqtt:subscribe(ConnPid, { + <<"$share/", Group/binary, "/foo/bar">>, 1 + }), + + Loop = fun(Self) -> + receive + wait -> + case last_message(<<"hello1">>, [ConnPid], 6000) of + {true, _} -> + Parent ! {got, ClientId}; + _ -> + Parent ! nothing + end, + Self(Self); + stop -> + stop + end + end, + + Loop(Loop) + end, + + Receive = fun() -> + receive + {got, _} = Got -> Got; + nothing -> nothing + after 6000 -> + nothing + end + end, + + Subs = [erlang:spawn(fun() -> MkSub(ClientId) end) || ClientId <- [ClientId1, ClientId2]], + timer:sleep(500), + + 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), + + [Sub ! stop || Sub <- Subs], + ok. + %%-------------------------------------------------------------------- %% help functions %%--------------------------------------------------------------------