fix(session): compatible to 4.3 takeover
This commit is contained in:
parent
0d7ad25804
commit
21ed258d58
|
@ -3,6 +3,7 @@
|
||||||
{VSN,
|
{VSN,
|
||||||
[{"4.4.1",
|
[{"4.4.1",
|
||||||
[{load_module,emqx_plugins,brutal_purge,soft_purge,[]},
|
[{load_module,emqx_plugins,brutal_purge,soft_purge,[]},
|
||||||
|
{load_module,emqx_session,brutal_purge,soft_purge,[]},
|
||||||
{load_module,emqx_app,brutal_purge,soft_purge,[]},
|
{load_module,emqx_app,brutal_purge,soft_purge,[]},
|
||||||
{add_module,emqx_relup},
|
{add_module,emqx_relup},
|
||||||
{load_module,emqx_os_mon,brutal_purge,soft_purge,[]},
|
{load_module,emqx_os_mon,brutal_purge,soft_purge,[]},
|
||||||
|
@ -40,6 +41,7 @@
|
||||||
{<<".*">>,[]}],
|
{<<".*">>,[]}],
|
||||||
[{"4.4.1",
|
[{"4.4.1",
|
||||||
[{load_module,emqx_plugins,brutal_purge,soft_purge,[]},
|
[{load_module,emqx_plugins,brutal_purge,soft_purge,[]},
|
||||||
|
{load_module,emqx_session,brutal_purge,soft_purge,[]},
|
||||||
{load_module,emqx_app,brutal_purge,soft_purge,[]},
|
{load_module,emqx_app,brutal_purge,soft_purge,[]},
|
||||||
{delete_module,emqx_relup},
|
{delete_module,emqx_relup},
|
||||||
{load_module,emqx_os_mon,brutal_purge,soft_purge,[]},
|
{load_module,emqx_os_mon,brutal_purge,soft_purge,[]},
|
||||||
|
|
|
@ -242,7 +242,8 @@ open_session(false, ClientInfo = #{clientid := ClientId}, ConnInfo) ->
|
||||||
{ok, #{session => Session, present => false}}
|
{ok, #{session => Session, present => false}}
|
||||||
end,
|
end,
|
||||||
case takeover_session(ClientId) of
|
case takeover_session(ClientId) of
|
||||||
{ok, ConnMod, ChanPid, Session} ->
|
{ok, ConnMod, ChanPid, Session0} ->
|
||||||
|
Session = emqx_session:upgrade(ClientInfo, Session0),
|
||||||
ok = emqx_session:resume(ClientInfo, Session),
|
ok = emqx_session:resume(ClientInfo, Session),
|
||||||
case request_stepdown({takeover, 'end'}, ConnMod, ChanPid) of
|
case request_stepdown({takeover, 'end'}, ConnMod, ChanPid) of
|
||||||
{ok, Pendings} ->
|
{ok, Pendings} ->
|
||||||
|
@ -289,7 +290,7 @@ takeover_session(ClientId, ChanPid) when node(ChanPid) == node() ->
|
||||||
ConnMod when is_atom(ConnMod) ->
|
ConnMod when is_atom(ConnMod) ->
|
||||||
case request_stepdown({takeover, 'begin'}, ConnMod, ChanPid) of
|
case request_stepdown({takeover, 'begin'}, ConnMod, ChanPid) of
|
||||||
{ok, Session} ->
|
{ok, Session} ->
|
||||||
{ok, ConnMod, ChanPid, Session};
|
{ok, ConnMod, ChanPid, emqx_session:downgrade(Session)};
|
||||||
{error, Reason} ->
|
{error, Reason} ->
|
||||||
{error, Reason}
|
{error, Reason}
|
||||||
end
|
end
|
||||||
|
|
|
@ -83,6 +83,8 @@
|
||||||
-export([ takeover/1
|
-export([ takeover/1
|
||||||
, resume/2
|
, resume/2
|
||||||
, replay/2
|
, replay/2
|
||||||
|
, upgrade/2
|
||||||
|
, downgrade/1
|
||||||
]).
|
]).
|
||||||
|
|
||||||
-export([expire/2]).
|
-export([expire/2]).
|
||||||
|
@ -196,6 +198,16 @@ init_mqueue(Zone) ->
|
||||||
default_priority => get_env(Zone, mqueue_default_priority, lowest)
|
default_priority => get_env(Zone, mqueue_default_priority, lowest)
|
||||||
}).
|
}).
|
||||||
|
|
||||||
|
%% @doc uprade from 4.3
|
||||||
|
upgrade(CInfo, S) ->
|
||||||
|
[session | Fields] = tuple_to_list(S),
|
||||||
|
#session{} = list_to_tuple([session, ?GET_CLIENT_ID(CInfo) | Fields] ++ [#{}]).
|
||||||
|
|
||||||
|
%% @doc Downgrade to 4.3
|
||||||
|
downgrade(#session{} = S) ->
|
||||||
|
[session, _ClientID | Fields] = tuple_to_list(S),
|
||||||
|
list_to_tuple([session | lists:reverse(tl(lists:reverse(Fields)))]).
|
||||||
|
|
||||||
%%--------------------------------------------------------------------
|
%%--------------------------------------------------------------------
|
||||||
%% Info, Stats
|
%% Info, Stats
|
||||||
%%--------------------------------------------------------------------
|
%%--------------------------------------------------------------------
|
||||||
|
|
Loading…
Reference in New Issue