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

chore(eviction): phase out session persistence concerns

This commit is contained in:
Andrew Mayorov 2023-07-18 15:47:12 +02:00
parent 0a00c39282
commit 3ba6d34511
No known key found for this signature in database
GPG Key ID: 2837C62ACFBFED5D
3 changed files with 3 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
apps/emqx_eviction_agent/src/emqx_eviction_agent.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_eviction_agent, [ {application, emqx_eviction_agent, [
{description, "EMQX Eviction Agent"}, {description, "EMQX Eviction Agent"},
{vsn, "5.0.1"}, {vsn, "5.1.0"},
{registered, [ {registered, [
emqx_eviction_agent_sup, emqx_eviction_agent_sup,
emqx_eviction_agent, emqx_eviction_agent,

12
apps/emqx_eviction_agent/src/emqx_eviction_agent_channel.erl
View File

@ -165,9 +165,8 @@ handle_cast(Msg, Channel) ->
?SLOG(error, #{msg => "unexpected_cast", cast => Msg}), ?SLOG(error, #{msg => "unexpected_cast", cast => Msg}),
{noreply, Channel}. {noreply, Channel}.
terminate(Reason, #{conninfo := ConnInfo, clientinfo := ClientInfo, session := Session} = Channel) -> terminate(Reason, #{clientinfo := ClientInfo, session := Session} = Channel) ->
ok = cancel_expiry_timer(Channel), ok = cancel_expiry_timer(Channel),
(Reason =:= expired) andalso emqx_persistent_session:persist(ClientInfo, ConnInfo, Session),
emqx_session:terminate(ClientInfo, Reason, Session). emqx_session:terminate(ClientInfo, Reason, Session).
code_change(_OldVsn, Channel, _Extra) -> code_change(_OldVsn, Channel, _Extra) ->
@ -205,10 +204,7 @@ handle_deliver(
Delivers1 = emqx_channel:maybe_nack(Delivers), Delivers1 = emqx_channel:maybe_nack(Delivers),
Delivers2 = emqx_session:ignore_local(ClientInfo, Delivers1, ClientId, Session), Delivers2 = emqx_session:ignore_local(ClientInfo, Delivers1, ClientId, Session),
NSession = emqx_session:enqueue(ClientInfo, Delivers2, Session), NSession = emqx_session:enqueue(ClientInfo, Delivers2, Session),
NChannel = persist(NSession, Channel), Channel#{session := NSession}.
%% We consider queued/dropped messages as delivered since they are now in the session state.
emqx_channel:maybe_mark_as_delivered(Session, Delivers),
NChannel.
cancel_expiry_timer(#{expiry_timer := TRef}) when is_reference(TRef) -> cancel_expiry_timer(#{expiry_timer := TRef}) when is_reference(TRef) ->
_ = erlang:cancel_timer(TRef), _ = erlang:cancel_timer(TRef),
@ -334,10 +330,6 @@ channel(ConnInfo, ClientInfo) ->
pendings => [] pendings => []
}. }.
persist(Session, #{clientinfo := ClientInfo, conninfo := ConnInfo} = Channel) ->
Session1 = emqx_persistent_session:persist(ClientInfo, ConnInfo, Session),
Channel#{session => Session1}.
info(Channel) -> info(Channel) ->
#{ #{
conninfo => maps:get(conninfo, Channel, undefined), conninfo => maps:get(conninfo, Channel, undefined),

3
apps/emqx_eviction_agent/test/emqx_eviction_agent_channel_SUITE.erl
View File

@ -36,9 +36,6 @@ init_per_testcase(_TestCase, Config) ->
Config. Config.
end_per_testcase(t_persistence, Config) -> end_per_testcase(t_persistence, Config) ->
emqx_config:put([persistent_session_store, enabled], false),
emqx_persistent_session:init_db_backend(),
?assertNot(emqx_persistent_session:is_store_enabled()),
Config; Config;
end_per_testcase(_TestCase, _Config) -> end_per_testcase(_TestCase, _Config) ->
ok. ok.