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

Merge pull request #6042 from emqx/persistent-session-review-follow-ups 

Persistent session review follow ups
This commit is contained in:
Tobias Lindahl 2021-11-02 10:48:33 +01:00 committed by GitHub
parent 62f9f6fd91 89cd68d36f
commit 7e8053248e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 7 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
apps/emqx/src/emqx_channel.erl
View File

@ -1179,14 +1179,13 @@ terminate(_, #channel{conn_state = idle}) -> ok;
terminate(normal, Channel) -> terminate(normal, Channel) ->
run_terminate_hook(normal, Channel); run_terminate_hook(normal, Channel);
terminate({shutdown, kicked}, Channel) -> terminate({shutdown, kicked}, Channel) ->
persist_if_session(Channel),
run_terminate_hook(kicked, Channel); run_terminate_hook(kicked, Channel);
terminate({shutdown, Reason}, Channel) when Reason =:= discarded; terminate({shutdown, Reason}, Channel) when Reason =:= discarded;
Reason =:= takeovered -> Reason =:= takeovered ->
run_terminate_hook(Reason, Channel); run_terminate_hook(Reason, Channel);
terminate(Reason, Channel = #channel{will_msg = WillMsg}) -> terminate(Reason, Channel = #channel{will_msg = WillMsg}) ->
(WillMsg =/= undefined) andalso publish_will_msg(WillMsg), (WillMsg =/= undefined) andalso publish_will_msg(WillMsg),
persist_if_session(Channel), (Reason =:= expired) andalso persist_if_session(Channel),
run_terminate_hook(Reason, Channel). run_terminate_hook(Reason, Channel).
persist_if_session(#channel{session = Session} = Channel) -> persist_if_session(#channel{session = Session} = Channel) ->

4
apps/emqx/src/emqx_cm.erl
View File

@ -58,6 +58,7 @@
, lookup_channels/2 , lookup_channels/2
]). ]).
%% Test/debug interface
-export([ all_channels/0 -export([ all_channels/0
, all_client_ids/0 , all_client_ids/0
]). ]).
@ -397,11 +398,12 @@ with_channel(ClientId, Fun) ->
Pids -> Fun(lists:last(Pids)) Pids -> Fun(lists:last(Pids))
end. end.
%% @doc Get all channels registed. %% @doc Get all registed channel pids. Debugg/test interface
all_channels() -> all_channels() ->
Pat = [{{'_', '$1'}, [], ['$1']}], Pat = [{{'_', '$1'}, [], ['$1']}],
ets:select(?CHAN_TAB, Pat). ets:select(?CHAN_TAB, Pat).
%% @doc Get all registed clientIDs. Debugg/test interface
all_client_ids() -> all_client_ids() ->
Pat = [{{'$1', '_'}, [], ['$1']}], Pat = [{{'$1', '_'}, [], ['$1']}],
ets:select(?CHAN_TAB, Pat). ets:select(?CHAN_TAB, Pat).

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

@ -113,8 +113,9 @@ init_per_group(snabbkaffe, Config) ->
[ {kill_connection_process, true} | Config]; [ {kill_connection_process, true} | Config];
init_per_group(gc_tests, Config) -> init_per_group(gc_tests, Config) ->
%% We need to make sure the system does not interfere with this test group. %% We need to make sure the system does not interfere with this test group.
[maybe_kill_connection_process(ClientId, [{kill_connection_process, true}]) lists:foreach(fun(ClientId) ->
|| ClientId <- emqx_cm:all_client_ids()], maybe_kill_connection_process(ClientId, [{kill_connection_process, true}])
end, emqx_cm:all_client_ids()),
emqx_common_test_helpers:stop_apps([]), emqx_common_test_helpers:stop_apps([]),
SessionMsgEts = gc_tests_session_store, SessionMsgEts = gc_tests_session_store,
MsgEts = gc_tests_msg_store, MsgEts = gc_tests_msg_store,