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

chore(persistent_sessions): address review comments

This commit is contained in:
Tobias Lindahl 2021-10-26 09:53:18 +02:00
parent 64787f4ccd
commit f39ccfb304
3 changed files with 5 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
apps/emqx/include/emqx.hrl
View File

@ -89,7 +89,7 @@
-record(route, { -record(route, {
topic :: binary(), topic :: binary(),
dest :: node() | {binary(), node()} | binary() dest :: node() | {binary(), node()} | emqx_session:sessionID()
}). }).
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------

7
apps/emqx/src/emqx_persistent_session_gc.erl
View File

@ -37,6 +37,9 @@
-endif. -endif.
-define(SERVER, ?MODULE). -define(SERVER, ?MODULE).
%% TODO: Maybe these should be configurable?
-define(MARKER_GRACE_PERIOD, 60000000).
-define(ABANDONED_GRACE_PERIOD, 300000000).
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% API %% API
@ -97,10 +100,6 @@ session_gc_timeout(_Ref, State) ->
session_gc_worker() -> session_gc_worker() ->
ok = emqx_persistent_session:gc_session_messages(fun session_gc_worker/2). ok = emqx_persistent_session:gc_session_messages(fun session_gc_worker/2).
%% TODO: Maybe these should be configurable?
-define(MARKER_GRACE_PERIOD, 60000000).
-define(ABANDONED_GRACE_PERIOD, 300000000).
session_gc_worker(delete, Key) -> session_gc_worker(delete, Key) ->
emqx_persistent_session:delete_session_message(Key); emqx_persistent_session:delete_session_message(Key);
session_gc_worker(marker, Key) -> session_gc_worker(marker, Key) ->

2
apps/emqx/src/emqx_session.erl
View File

@ -259,7 +259,7 @@ stats(Session) -> info(?STATS_KEYS, Session).
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
ignore_local(Delivers, Subscriber, Session) -> ignore_local(Delivers, Subscriber, Session) ->
Subs = emqx_session:info(subscriptions, Session), Subs = info(subscriptions, Session),
lists:dropwhile(fun({deliver, Topic, #message{from = Publisher}}) -> lists:dropwhile(fun({deliver, Topic, #message{from = Publisher}}) ->
case maps:find(Topic, Subs) of case maps:find(Topic, Subs) of
{ok, #{nl := 1}} when Subscriber =:= Publisher -> {ok, #{nl := 1}} when Subscriber =:= Publisher ->