Fix issue#1833 and #1834 (#1845)

This commit is contained in:
tigercl 2018-09-22 16:10:24 +08:00 committed by Shawn
parent 8653732bae
commit 03f607c1b2
5 changed files with 58 additions and 35 deletions

View File

@ -797,7 +797,7 @@ end}.
%% @doc Session Expiry Interval %% @doc Session Expiry Interval
{mapping, "zone.$name.session_expiry_interval", "emqx.zones", [ {mapping, "zone.$name.session_expiry_interval", "emqx.zones", [
{default, "2h"}, {default, "2h"},
{datatype, {duration, ms}} {datatype, {duration, s}}
]}. ]}.
%% @doc Type: simple | priority %% @doc Type: simple | priority

View File

@ -410,9 +410,17 @@ process_packet(?UNSUBSCRIBE_PACKET(PacketId, Properties, RawTopicFilters),
process_packet(?PACKET(?PINGREQ), PState) -> process_packet(?PACKET(?PINGREQ), PState) ->
send(?PACKET(?PINGRESP), PState); send(?PACKET(?PINGRESP), PState);
process_packet(?DISCONNECT_PACKET(?RC_SUCCESS), PState) -> process_packet(?DISCONNECT_PACKET(?RC_SUCCESS, #{'Session-Expiry-Interval' := Interval}),
%% Clean willmsg PState = #pstate{session = SPid, conn_props = #{'Session-Expiry-Interval' := OldInterval}}) ->
{stop, normal, PState#pstate{will_msg = undefined}}; case Interval =/= 0 andalso OldInterval =:= 0 of
true ->
deliver({disconnect, ?RC_PROTOCOL_ERROR}, PState),
{error, protocol_error, PState};
false ->
emqx_session:update_expiry_interval(SPid, Interval),
%% Clean willmsg
{stop, normal, PState#pstate{will_msg = undefined}}
end;
process_packet(?DISCONNECT_PACKET(_), PState) -> process_packet(?DISCONNECT_PACKET(_), PState) ->
{stop, normal, PState}. {stop, normal, PState}.
@ -562,17 +570,32 @@ maybe_assign_client_id(PState) ->
PState. PState.
try_open_session(#pstate{zone = Zone, try_open_session(#pstate{zone = Zone,
proto_ver = ProtoVer,
client_id = ClientId, client_id = ClientId,
conn_pid = ConnPid, conn_pid = ConnPid,
conn_props = ConnProps, conn_props = ConnProps,
username = Username, username = Username,
clean_start = CleanStart}) -> clean_start = CleanStart}) ->
case emqx_sm:open_session(#{zone => Zone,
client_id => ClientId, SessAttrs = #{
conn_pid => ConnPid, zone => Zone,
username => Username, client_id => ClientId,
clean_start => CleanStart, conn_pid => ConnPid,
conn_props => ConnProps}) of username => Username,
clean_start => CleanStart
},
case emqx_sm:open_session(maps:put(expiry_interval, if
ProtoVer =:= ?MQTT_PROTO_V5 ->
maps:get('Session-Expiry-Interval', ConnProps, 0);
true ->
case CleanStart of
true ->
0;
false ->
emqx_zone:get_env(Zone, session_expiry_interval, 16#ffffffff)
end
end, SessAttrs)) of
{ok, SPid} -> {ok, SPid} ->
{ok, SPid, false}; {ok, SPid, false};
Other -> Other Other -> Other

View File

@ -47,6 +47,7 @@
-export([info/1, attrs/1]). -export([info/1, attrs/1]).
-export([stats/1]). -export([stats/1]).
-export([resume/2, discard/2]). -export([resume/2, discard/2]).
-export([update_expiry_interval/2]).
-export([subscribe/2, subscribe/4]). -export([subscribe/2, subscribe/4]).
-export([publish/3]). -export([publish/3]).
-export([puback/2, puback/3]). -export([puback/2, puback/3]).
@ -313,6 +314,10 @@ resume(SPid, ConnPid) ->
discard(SPid, ByPid) -> discard(SPid, ByPid) ->
gen_server:call(SPid, {discard, ByPid}, infinity). gen_server:call(SPid, {discard, ByPid}, infinity).
-spec(update_expiry_interval(spid(), timeout()) -> ok).
update_expiry_interval(SPid, Interval) ->
gen_server:cast(SPid, {expiry_interval, Interval * 1000}).
-spec(close(spid()) -> ok). -spec(close(spid()) -> ok).
close(SPid) -> close(SPid) ->
gen_server:call(SPid, close, infinity). gen_server:call(SPid, close, infinity).
@ -321,12 +326,12 @@ close(SPid) ->
%% gen_server callbacks %% gen_server callbacks
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
init([Parent, #{zone := Zone, init([Parent, #{zone := Zone,
client_id := ClientId, client_id := ClientId,
username := Username, username := Username,
conn_pid := ConnPid, conn_pid := ConnPid,
clean_start := CleanStart, clean_start := CleanStart,
conn_props := ConnProps}]) -> expiry_interval := ExpiryInterval}]) ->
process_flag(trap_exit, true), process_flag(trap_exit, true),
true = link(ConnPid), true = link(ConnPid),
MaxInflight = get_env(Zone, max_inflight), MaxInflight = get_env(Zone, max_inflight),
@ -346,7 +351,7 @@ init([Parent, #{zone := Zone,
awaiting_rel = #{}, awaiting_rel = #{},
await_rel_timeout = get_env(Zone, await_rel_timeout), await_rel_timeout = get_env(Zone, await_rel_timeout),
max_awaiting_rel = get_env(Zone, max_awaiting_rel), max_awaiting_rel = get_env(Zone, max_awaiting_rel),
expiry_interval = expire_interval(Zone, ConnProps), expiry_interval = ExpiryInterval,
enable_stats = get_env(Zone, enable_stats, true), enable_stats = get_env(Zone, enable_stats, true),
deliver_stats = 0, deliver_stats = 0,
enqueue_stats = 0, enqueue_stats = 0,
@ -361,11 +366,6 @@ init([Parent, #{zone := Zone,
ok = proc_lib:init_ack(Parent, {ok, self()}), ok = proc_lib:init_ack(Parent, {ok, self()}),
gen_server:enter_loop(?MODULE, [{hibernate_after, IdleTimout}], State). gen_server:enter_loop(?MODULE, [{hibernate_after, IdleTimout}], State).
expire_interval(_Zone, #{'Session-Expiry-Interval' := I}) ->
I * 1000;
expire_interval(Zone, _ConnProps) -> %% Maybe v3.1.1
get_env(Zone, session_expiry_interval, 0).
init_mqueue(Zone) -> init_mqueue(Zone) ->
emqx_mqueue:init(#{type => get_env(Zone, mqueue_type, simple), emqx_mqueue:init(#{type => get_env(Zone, mqueue_type, simple),
max_len => get_env(Zone, max_mqueue_len, 1000), max_len => get_env(Zone, max_mqueue_len, 1000),
@ -540,6 +540,9 @@ handle_cast({resume, ConnPid}, State = #state{client_id = ClientId,
%% Replay delivery and Dequeue pending messages %% Replay delivery and Dequeue pending messages
noreply(dequeue(retry_delivery(true, State1))); noreply(dequeue(retry_delivery(true, State1)));
handle_cast({expiry_interval, Interval}, State) ->
{noreply, State#state{expiry_interval = Interval}};
handle_cast(Msg, State) -> handle_cast(Msg, State) ->
emqx_logger:error("[Session] unexpected cast: ~p", [Msg]), emqx_logger:error("[Session] unexpected cast: ~p", [Msg]),
{noreply, State}. {noreply, State}.
@ -591,13 +594,10 @@ handle_info({timeout, Timer, expired}, State = #state{expiry_timer = Timer}) ->
?LOG(info, "expired, shutdown now:(", [], State), ?LOG(info, "expired, shutdown now:(", [], State),
shutdown(expired, State); shutdown(expired, State);
handle_info({'EXIT', ConnPid, Reason}, State = #state{clean_start = true, conn_pid = ConnPid}) ->
{stop, Reason, State#state{conn_pid = undefined}};
handle_info({'EXIT', ConnPid, Reason}, State = #state{expiry_interval = 0, conn_pid = ConnPid}) -> handle_info({'EXIT', ConnPid, Reason}, State = #state{expiry_interval = 0, conn_pid = ConnPid}) ->
{stop, Reason, State#state{conn_pid = undefined}}; {stop, Reason, State#state{conn_pid = undefined}};
handle_info({'EXIT', ConnPid, _Reason}, State = #state{clean_start = false, conn_pid = ConnPid}) -> handle_info({'EXIT', ConnPid, _Reason}, State = #state{conn_pid = ConnPid}) ->
{noreply, ensure_expire_timer(State#state{conn_pid = undefined})}; {noreply, ensure_expire_timer(State#state{conn_pid = undefined})};
handle_info({'EXIT', OldPid, _Reason}, State = #state{old_conn_pid = OldPid}) -> handle_info({'EXIT', OldPid, _Reason}, State = #state{old_conn_pid = OldPid}) ->
@ -876,8 +876,8 @@ ensure_retry_timer(Interval, State = #state{retry_timer = undefined}) ->
ensure_retry_timer(_Timeout, State) -> ensure_retry_timer(_Timeout, State) ->
State. State.
ensure_expire_timer(State = #state{expiry_interval = Interval}) when Interval > 0 -> ensure_expire_timer(State = #state{expiry_interval = Interval}) when Interval > 0 andalso Interval =/= 16#ffffffff ->
State#state{expiry_timer = emqx_misc:start_timer(Interval, expired)}; State#state{expiry_timer = emqx_misc:start_timer(Interval * 1000, expired)};
ensure_expire_timer(State) -> ensure_expire_timer(State) ->
State. State.

View File

@ -53,12 +53,12 @@ init([ClientId]) ->
}. }.
handle_call({start_session, ClientPid, ClientId, Zone}, _From, State) -> handle_call({start_session, ClientPid, ClientId, Zone}, _From, State) ->
Attrs = #{ zone => Zone, Attrs = #{ zone => Zone,
client_id => ClientId, client_id => ClientId,
conn_pid => ClientPid, conn_pid => ClientPid,
clean_start => true, clean_start => true,
username => undefined, username => undefined,
conn_props => undefined expiry_interval => 0
}, },
{ok, SessPid} = emqx_sm:open_session(Attrs), {ok, SessPid} = emqx_sm:open_session(Attrs),
{reply, {ok, SessPid}, State#state{ {reply, {ok, SessPid}, State#state{

View File

@ -25,7 +25,7 @@ t_open_close_session(_) ->
emqx_ct_broker_helpers:run_setup_steps(), emqx_ct_broker_helpers:run_setup_steps(),
{ok, ClientPid} = emqx_mock_client:start_link(<<"client">>), {ok, ClientPid} = emqx_mock_client:start_link(<<"client">>),
Attrs = #{clean_start => true, client_id => <<"client">>, conn_pid => ClientPid, Attrs = #{clean_start => true, client_id => <<"client">>, conn_pid => ClientPid,
zone => internal, username => <<"zhou">>, conn_props => #{}}, zone => internal, username => <<"zhou">>, expiry_interval => 0},
{ok, SPid} = emqx_sm:open_session(Attrs), {ok, SPid} = emqx_sm:open_session(Attrs),
[{<<"client">>, SPid}] = emqx_sm:lookup_session(<<"client">>), [{<<"client">>, SPid}] = emqx_sm:lookup_session(<<"client">>),
SPid = emqx_sm:lookup_session_pid(<<"client">>), SPid = emqx_sm:lookup_session_pid(<<"client">>),