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

refactor log

This commit is contained in:
Feng 2015-10-30 21:20:37 +08:00
parent 8d3b7b4c24
commit 4ed62d018e
4 changed files with 136 additions and 133 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
src/emqttd_cm.erl
View File

@ -37,8 +37,6 @@
-behaviour(gen_server2). -behaviour(gen_server2).
-define(SERVER, ?MODULE).
%% gen_server Function Exports %% gen_server Function Exports
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
terminate/2, code_change/3]). terminate/2, code_change/3]).
@ -47,6 +45,9 @@
-define(CM_POOL, ?MODULE). -define(CM_POOL, ?MODULE).
-define(LOG(Level, Format, Args, Client),
lager:Level("CM(~s): " ++ Format, [Client#mqtt_client.client_id|Args])).
%%%============================================================================= %%%=============================================================================
%%% API %%% API
%%%============================================================================= %%%=============================================================================
@ -102,15 +103,16 @@ init([Id, StatsFun]) ->
handle_call(Req, _From, State) -> handle_call(Req, _From, State) ->
lager:error("unexpected request: ~p", [Req]), lager:error("unexpected request: ~p", [Req]),
{reply, {error, badreq}, State}. {reply, {error, unsupported_req}, State}.
handle_cast({register, Client = #mqtt_client{client_id = ClientId, client_pid = Pid}}, State) -> handle_cast({register, Client = #mqtt_client{client_id = ClientId,
client_pid = Pid}}, State) ->
case ets:lookup(mqtt_client, ClientId) of case ets:lookup(mqtt_client, ClientId) of
[#mqtt_client{client_pid = Pid}] -> [#mqtt_client{client_pid = Pid}] ->
lager:error("ClientId '~s' has been registered with ~p", [ClientId, Pid]),
ignore; ignore;
[#mqtt_client{client_pid = OldPid}] -> [#mqtt_client{client_pid = OldPid}] ->
lager:warning("ClientId '~s' is duplicated: pid=~p, oldpid=~p", [ClientId, Pid, OldPid]); %% TODO: should cancel monitor
?LOG(warning, "client ~p conflict with ~p", [Pid, OldPid], Client);
[] -> [] ->
ok ok
end, end,
@ -121,10 +123,10 @@ handle_cast({unregister, ClientId, Pid}, State) ->
case ets:lookup(mqtt_client, ClientId) of case ets:lookup(mqtt_client, ClientId) of
[#mqtt_client{client_pid = Pid}] -> [#mqtt_client{client_pid = Pid}] ->
ets:delete(mqtt_client, ClientId); ets:delete(mqtt_client, ClientId);
[_] -> [_] ->
ignore; ignore;
[] -> [] ->
lager:error("Cannot find clientId '~s' with ~p", [ClientId, Pid]) ?LOG(error, "Cannot find registered: ~p", [Pid], State)
end, end,
{noreply, setstats(State)}; {noreply, setstats(State)};
@ -137,7 +139,8 @@ handle_info(Info, State) ->
{noreply, State}. {noreply, State}.
terminate(_Reason, #state{id = Id}) -> terminate(_Reason, #state{id = Id}) ->
gproc_pool:disconnect_worker(?CM_POOL, {?MODULE, Id}), ok. gproc_pool:disconnect_worker(?CM_POOL, {?MODULE, Id}),
ok.
code_change(_OldVsn, State, _Extra) -> code_change(_OldVsn, State, _Extra) ->
{ok, State}. {ok, State}.

4
src/emqttd_pubsub.erl
View File

@ -24,7 +24,6 @@
%%% %%%
%%% @end %%% @end
%%%----------------------------------------------------------------------------- %%%-----------------------------------------------------------------------------
-module(emqttd_pubsub). -module(emqttd_pubsub).
-author("Feng Lee <feng@emqtt.io>"). -author("Feng Lee <feng@emqtt.io>").
@ -48,8 +47,7 @@
publish/1]). publish/1]).
%% Local node %% Local node
-export([dispatch/2, -export([dispatch/2, match/1]).
match/1]).
-behaviour(gen_server2). -behaviour(gen_server2).

208
src/emqttd_session.erl
View File

@ -44,7 +44,6 @@
%%% %%%
%%% @end %%% @end
%%%----------------------------------------------------------------------------- %%%-----------------------------------------------------------------------------
-module(emqttd_session). -module(emqttd_session).
-author("Feng Lee <feng@emqtt.io>"). -author("Feng Lee <feng@emqtt.io>").
@ -53,16 +52,15 @@
-include("emqttd_protocol.hrl"). -include("emqttd_protocol.hrl").
-behaviour(gen_server2).
%% Session API %% Session API
-export([start_link/3, resume/3, destroy/2]). -export([start_link/3, resume/3, info/1, destroy/2]).
%% PubSub APIs %% PubSub APIs
-export([publish/2, -export([publish/2, puback/2, pubrec/2, pubrel/2, pubcomp/2,
puback/2, pubrec/2, pubrel/2, pubcomp/2,
subscribe/2, subscribe/3, unsubscribe/2]). subscribe/2, subscribe/3, unsubscribe/2]).
-behaviour(gen_server2).
%% gen_server Function Exports %% gen_server Function Exports
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
terminate/2, code_change/3]). terminate/2, code_change/3]).
@ -135,6 +133,10 @@
-define(PUBSUB_TIMEOUT, 60000). -define(PUBSUB_TIMEOUT, 60000).
-define(LOG(Level, Format, Args, State),
lager:Level([{client, State#session.client_id}],
"Session(~s): " ++ Format, [State#session.client_id | Args])).
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
%% @doc Start a session. %% @doc Start a session.
%% @end %% @end
@ -151,6 +153,13 @@ start_link(CleanSess, ClientId, ClientPid) ->
resume(SessPid, ClientId, ClientPid) -> resume(SessPid, ClientId, ClientPid) ->
gen_server2:cast(SessPid, {resume, ClientId, ClientPid}). gen_server2:cast(SessPid, {resume, ClientId, ClientPid}).
%%------------------------------------------------------------------------------
%% @doc Session Info.
%% @end
%%------------------------------------------------------------------------------
info(SessPid) ->
gen_server2:call(SessPid, info).
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
%% @doc Destroy a session. %% @doc Destroy a session.
%% @end %% @end
@ -171,7 +180,7 @@ subscribe(SessPid, TopicTable) ->
subscribe(SessPid, PacketId, TopicTable) -> subscribe(SessPid, PacketId, TopicTable) ->
From = self(), From = self(),
AckFun = fun(GrantedQos) -> AckFun = fun(GrantedQos) ->
From ! {suback, PacketId, GrantedQos} From ! {suback, PacketId, GrantedQos}
end, end,
gen_server2:cast(SessPid, {subscribe, TopicTable, AckFun}). gen_server2:cast(SessPid, {subscribe, TopicTable, AckFun}).
@ -246,12 +255,15 @@ init([CleanSess, ClientId, ClientPid]) ->
expired_after = emqttd_opts:g(expired_after, SessEnv) * 3600, expired_after = emqttd_opts:g(expired_after, SessEnv) * 3600,
collect_interval = emqttd_opts:g(collect_interval, SessEnv, 0), collect_interval = emqttd_opts:g(collect_interval, SessEnv, 0),
timestamp = os:timestamp()}, timestamp = os:timestamp()},
emqttd_sm:register_session(CleanSess, ClientId, info(Session)), emqttd_sm:register_session(CleanSess, ClientId, sess_info(Session)),
%% start statistics %% start statistics
{ok, start_collector(Session), hibernate}. {ok, start_collector(Session), hibernate}.
prioritise_call(Msg, _From, _Len, _State) -> prioritise_call(Msg, _From, _Len, _State) ->
case Msg of _ -> 0 end. case Msg of
info -> 10;
_ -> 0
end.
prioritise_cast(Msg, _Len, _State) -> prioritise_cast(Msg, _Len, _State) ->
case Msg of case Msg of
@ -269,13 +281,16 @@ prioritise_cast(Msg, _Len, _State) ->
prioritise_info(Msg, _Len, _State) -> prioritise_info(Msg, _Len, _State) ->
case Msg of case Msg of
{'EXIT', _, _} -> 10; {'EXIT', _, _} -> 10;
session_expired -> 10; expired -> 10;
{timeout, _, _} -> 5; {timeout, _, _} -> 5;
collect_info -> 2; collect_info -> 2;
{dispatch, _} -> 1; {dispatch, _} -> 1;
_ -> 0 _ -> 0
end. end.
handle_call(info, _From, State) ->
{reply, sess_info(State), State};
handle_call({publish, Msg = #mqtt_message{qos = ?QOS_2, pktid = PktId}}, _From, handle_call({publish, Msg = #mqtt_message{qos = ?QOS_2, pktid = PktId}}, _From,
Session = #session{client_id = ClientId, Session = #session{client_id = ClientId,
awaiting_rel = AwaitingRel, awaiting_rel = AwaitingRel,
@ -292,11 +307,11 @@ handle_call({publish, Msg = #mqtt_message{qos = ?QOS_2, pktid = PktId}}, _From,
end; end;
handle_call(Req, _From, State) -> handle_call(Req, _From, State) ->
lager:error("Unexpected Request: ~p", [Req]), ?LOG(critical, "Unexpected Request: ~p", [Req], State),
{reply, ok, State}. {reply, {error, unsupported_req}, State}.
handle_cast({subscribe, TopicTable0, AckFun}, Session = #session{ handle_cast({subscribe, TopicTable0, AckFun}, Session = #session{client_id = ClientId,
client_id = ClientId, subscriptions = Subscriptions}) -> subscriptions = Subscriptions}) ->
TopicTable = emqttd_broker:foldl_hooks('client.subscribe', [ClientId], TopicTable0), TopicTable = emqttd_broker:foldl_hooks('client.subscribe', [ClientId], TopicTable0),
@ -312,18 +327,16 @@ handle_cast({subscribe, TopicTable0, AckFun}, Session = #session{
emqttd_broker:foreach_hooks('client.subscribe.after', [ClientId, TopicTable]), emqttd_broker:foreach_hooks('client.subscribe.after', [ClientId, TopicTable]),
lager:info([{client, ClientId}], "Session(~s): subscribe ~p, Granted QoS: ~p", ?LOG(info, "Subscribe ~p, Granted QoS: ~p", [TopicTable, GrantedQos], Session),
[ClientId, TopicTable, GrantedQos]),
Subscriptions1 = Subscriptions1 =
lists:foldl(fun({Topic, Qos}, Acc) -> lists:foldl(fun({Topic, Qos}, Acc) ->
case lists:keyfind(Topic, 1, Acc) of case lists:keyfind(Topic, 1, Acc) of
{Topic, Qos} -> {Topic, Qos} ->
lager:warning([{client, ClientId}], "Session(~s): " ?LOG(warning, "resubscribe ~s, qos = ~w", [Topic, Qos], Session),
"resubscribe ~s, qos = ~w", [ClientId, Topic, Qos]), Acc; Acc;
{Topic, OldQos} -> {Topic, OldQos} ->
lager:warning([{client, ClientId}], "Session(~s): " ?LOG(warning, "resubscribe ~s, old qos=~w, new qos=~w", [Topic, OldQos, Qos], Session),
"resubscribe ~s, old qos=~w, new qos=~w", [ClientId, Topic, OldQos, Qos]),
lists:keyreplace(Topic, 1, Acc, {Topic, Qos}); lists:keyreplace(Topic, 1, Acc, {Topic, Qos});
false -> false ->
%%TODO: the design is ugly, rewrite later...:( %%TODO: the design is ugly, rewrite later...:(
@ -353,41 +366,31 @@ handle_cast({unsubscribe, Topics0}, Session = #session{client_id = ClientId,
{Topic, _Qos} -> {Topic, _Qos} ->
lists:keydelete(Topic, 1, Acc); lists:keydelete(Topic, 1, Acc);
false -> false ->
lager:warning([{client, ClientId}], "Session(~s) not subscribe ~s", [ClientId, Topic]), Acc Acc
end end
end, Subscriptions, Topics), end, Subscriptions, Topics),
noreply(Session#session{subscriptions = Subscriptions1}); noreply(Session#session{subscriptions = Subscriptions1});
handle_cast({destroy, ClientId}, Session = #session{client_id = ClientId}) -> handle_cast({destroy, ClientId}, Session = #session{client_id = ClientId}) ->
lager:warning([{client, ClientId}], "Session(~s) destroyed", [ClientId]), ?LOG(warning, "destroyed", [], Session),
{stop, {shutdown, destroy}, Session}; {stop, {shutdown, destroy}, Session};
handle_cast({resume, ClientId, ClientPid}, Session) -> handle_cast({resume, ClientId, ClientPid}, Session = #session{client_id = ClientId,
client_pid = OldClientPid,
inflight_queue = InflightQ,
awaiting_ack = AwaitingAck,
awaiting_comp = AwaitingComp,
expired_timer = ETimer} = Session) ->
#session{client_id = ClientId, ?LOG(info, "resumed by ~p", [ClientPid], Session),
client_pid = OldClientPid,
inflight_queue = InflightQ,
awaiting_ack = AwaitingAck,
awaiting_comp = AwaitingComp,
expired_timer = ETimer} = Session,
lager:info([{client, ClientId}], "Session(~s) resumed by ~p", [ClientId, ClientPid]), %% Cancel expired timer
%% cancel expired timer
cancel_timer(ETimer), cancel_timer(ETimer),
%% Kickout old client case kick(ClientId, OldClientPid, ClientPid) of
if ok -> ?LOG(warning, "~p kickout ~p", [ClientPid, OldClientPid], Session);
OldClientPid == undefined -> ignore -> ok
ok;
OldClientPid == ClientPid ->
ok; %% ??
true ->
lager:error([{client, ClientId}], "Session(~s): ~p kickout ~p",
[ClientId, ClientPid, OldClientPid]),
unlink(OldClientPid),
OldClientPid ! {stop, duplicate_id, ClientPid}
end, end,
true = link(ClientPid), true = link(ClientPid),
@ -416,19 +419,18 @@ handle_cast({resume, ClientId, ClientPid}, Session) ->
noreply(dequeue(Session2)); noreply(dequeue(Session2));
%% PUBACK %% PUBACK
handle_cast({puback, PktId}, Session = #session{client_id = ClientId, awaiting_ack = AwaitingAck}) -> handle_cast({puback, PktId}, Session = #session{awaiting_ack = AwaitingAck}) ->
case maps:find(PktId, AwaitingAck) of case maps:find(PktId, AwaitingAck) of
{ok, TRef} -> {ok, TRef} ->
cancel_timer(TRef), cancel_timer(TRef),
noreply(dequeue(acked(PktId, Session))); noreply(dequeue(acked(PktId, Session)));
error -> error ->
lager:error([{client, ClientId}], "Session(~s) cannot find PUBACK ~w", [ClientId, PktId]), ?LOG(error, "Cannot find PUBACK: ~p", [PktId], Session),
noreply(Session) noreply(Session)
end; end;
%% PUBREC %% PUBREC
handle_cast({pubrec, PktId}, Session = #session{client_id = ClientId, handle_cast({pubrec, PktId}, Session = #session{awaiting_ack = AwaitingAck,
awaiting_ack = AwaitingAck,
awaiting_comp = AwaitingComp, awaiting_comp = AwaitingComp,
await_rel_timeout = Timeout}) -> await_rel_timeout = Timeout}) ->
case maps:find(PktId, AwaitingAck) of case maps:find(PktId, AwaitingAck) of
@ -439,37 +441,36 @@ handle_cast({pubrec, PktId}, Session = #session{client_id = ClientId,
Session1 = acked(PktId, Session#session{awaiting_comp = AwaitingComp1}), Session1 = acked(PktId, Session#session{awaiting_comp = AwaitingComp1}),
noreply(dequeue(Session1)); noreply(dequeue(Session1));
error -> error ->
lager:error([{client, ClientId}], "Session(~s) cannot find PUBREC ~w", [ClientId, PktId]), ?LOG(error, "Cannot find PUBREC: ~p", [PktId], Session),
noreply(Session) noreply(Session)
end; end;
%% PUBREL %% PUBREL
handle_cast({pubrel, PktId}, Session = #session{client_id = ClientId, handle_cast({pubrel, PktId}, Session = #session{awaiting_rel = AwaitingRel}) ->
awaiting_rel = AwaitingRel}) ->
case maps:find(PktId, AwaitingRel) of case maps:find(PktId, AwaitingRel) of
{ok, {Msg, TRef}} -> {ok, {Msg, TRef}} ->
cancel_timer(TRef), cancel_timer(TRef),
emqttd_pubsub:publish(Msg), emqttd_pubsub:publish(Msg),
noreply(Session#session{awaiting_rel = maps:remove(PktId, AwaitingRel)}); noreply(Session#session{awaiting_rel = maps:remove(PktId, AwaitingRel)});
error -> error ->
lager:error([{client, ClientId}], "Session(~s) cannot find PUBREL ~w", [ClientId, PktId]), ?LOG(error, "Cannot find PUBREL: ~p", [PktId], Session),
noreply(Session) noreply(Session)
end; end;
%% PUBCOMP %% PUBCOMP
handle_cast({pubcomp, PktId}, Session = #session{client_id = ClientId, awaiting_comp = AwaitingComp}) -> handle_cast({pubcomp, PktId}, Session = #session{awaiting_comp = AwaitingComp}) ->
case maps:find(PktId, AwaitingComp) of case maps:find(PktId, AwaitingComp) of
{ok, TRef} -> {ok, TRef} ->
cancel_timer(TRef), cancel_timer(TRef),
noreply(Session#session{awaiting_comp = maps:remove(PktId, AwaitingComp)}); noreply(Session#session{awaiting_comp = maps:remove(PktId, AwaitingComp)});
error -> error ->
lager:error("Session(~s) cannot find PUBCOMP ~w", [ClientId, PktId]), ?LOG(error, "Cannot find PUBCOMP: ~p", [PktId], Session),
noreply(Session) noreply(Session)
end; end;
handle_cast(Msg, State) -> handle_cast(Msg, State) ->
lager:error("Unexpected Msg: ~p, State: ~p", [Msg, State]), ?LOG(critical, "Unexpected Msg: ~p", [Msg], State),
{noreply, State}. noreply(State).
%% Queue messages when client is offline %% Queue messages when client is offline
handle_info({dispatch, Msg}, Session = #session{client_pid = undefined, handle_info({dispatch, Msg}, Session = #session{client_pid = undefined,
@ -483,14 +484,15 @@ handle_info({dispatch, Msg = #mqtt_message{qos = ?QOS_0}},
ClientPid ! {deliver, Msg}, ClientPid ! {deliver, Msg},
noreply(Session); noreply(Session);
handle_info({dispatch, Msg = #mqtt_message{qos = QoS}}, Session = #session{message_queue = MsgQ}) handle_info({dispatch, Msg = #mqtt_message{qos = QoS}},
Session = #session{message_queue = MsgQ})
when QoS =:= ?QOS_1 orelse QoS =:= ?QOS_2 -> when QoS =:= ?QOS_1 orelse QoS =:= ?QOS_2 ->
case check_inflight(Session) of case check_inflight(Session) of
true -> true ->
{noreply, deliver(Msg, Session)}; noreply(deliver(Msg, Session));
false -> false ->
{noreply, Session#session{message_queue = emqttd_mqueue:in(Msg, MsgQ)}} noreply(Session#session{message_queue = emqttd_mqueue:in(Msg, MsgQ)})
end; end;
handle_info({timeout, awaiting_ack, PktId}, Session = #session{client_pid = undefined, handle_info({timeout, awaiting_ack, PktId}, Session = #session{client_pid = undefined,
@ -498,81 +500,70 @@ handle_info({timeout, awaiting_ack, PktId}, Session = #session{client_pid = unde
%% just remove awaiting %% just remove awaiting
noreply(Session#session{awaiting_ack = maps:remove(PktId, AwaitingAck)}); noreply(Session#session{awaiting_ack = maps:remove(PktId, AwaitingAck)});
handle_info({timeout, awaiting_ack, PktId}, Session = #session{client_id = ClientId, handle_info({timeout, awaiting_ack, PktId}, Session = #session{inflight_queue = InflightQ,
inflight_queue = InflightQ,
awaiting_ack = AwaitingAck}) -> awaiting_ack = AwaitingAck}) ->
lager:info("Awaiting Ack Timeout: ~p:", [PktId]),
case maps:find(PktId, AwaitingAck) of case maps:find(PktId, AwaitingAck) of
{ok, _TRef} -> {ok, _TRef} ->
case lists:keyfind(PktId, 1, InflightQ) of case lists:keyfind(PktId, 1, InflightQ) of
{_, Msg} -> {_, Msg} ->
noreply(redeliver(Msg, Session)); noreply(redeliver(Msg, Session));
false -> false ->
lager:error([{client, ClientId}], "Session(~s):" ?LOG(error, "AwaitingAck timeout but Cannot find PktId: ~p", [PktId], Session),
"Awaiting timeout but Cannot find PktId :~p", [ClientId, PktId]),
noreply(dequeue(Session)) noreply(dequeue(Session))
end; end;
error -> error ->
lager:error([{client, ClientId}], "Session(~s):" ?LOG(error, "Cannot find AwaitingAck: ~p", [PktId], Session),
"Cannot find Awaiting Ack:~p", [ClientId, PktId]),
noreply(Session) noreply(Session)
end; end;
handle_info({timeout, awaiting_rel, PktId}, Session = #session{client_id = ClientId, handle_info({timeout, awaiting_rel, PktId}, Session = #session{awaiting_rel = AwaitingRel}) ->
awaiting_rel = AwaitingRel}) ->
case maps:find(PktId, AwaitingRel) of case maps:find(PktId, AwaitingRel) of
{ok, {Msg, _TRef}} -> {ok, {_Msg, _TRef}} ->
lager:error([{client, ClientId}], "Session(~s) AwaitingRel Timout!~n" ?LOG(error, "AwaitingRel Timout: ~p, Drop Message!", [PktId], Session),
"Drop Message:~p", [ClientId, Msg]),
noreply(Session#session{awaiting_rel = maps:remove(PktId, AwaitingRel)}); noreply(Session#session{awaiting_rel = maps:remove(PktId, AwaitingRel)});
error -> error ->
lager:error([{client, ClientId}], "Session(~s) cannot find AwaitingRel ~w", [ClientId, PktId]), ?LOG(error, "Cannot find AwaitingRel: ~p", [PktId], Session),
{noreply, Session, hibernate} noreply(Session)
end; end;
handle_info({timeout, awaiting_comp, PktId}, Session = #session{client_id = ClientId, handle_info({timeout, awaiting_comp, PktId}, Session = #session{awaiting_comp = Awaiting}) ->
awaiting_comp = Awaiting}) ->
case maps:find(PktId, Awaiting) of case maps:find(PktId, Awaiting) of
{ok, _TRef} -> {ok, _TRef} ->
lager:error([{client, ClientId}], "Session(~s) " ?LOG(error, "Awaiting PUBCOMP Timout: ~p", [PktId], Session),
"Awaiting PUBCOMP Timout: PktId=~p!", [ClientId, PktId]),
noreply(Session#session{awaiting_comp = maps:remove(PktId, Awaiting)}); noreply(Session#session{awaiting_comp = maps:remove(PktId, Awaiting)});
error -> error ->
lager:error([{client, ClientId}], "Session(~s) " ?LOG(error, "Cannot find Awaiting PUBCOMP: ~p", [PktId], Session),
"Cannot find Awaiting PUBCOMP: PktId=~p", [ClientId, PktId]),
noreply(Session) noreply(Session)
end; end;
handle_info(collect_info, Session = #session{clean_sess = CleanSess, client_id = ClientId}) -> handle_info(collect_info, Session = #session{clean_sess = CleanSess, client_id = ClientId}) ->
emqttd_sm:register_session(CleanSess, ClientId, info(Session)), emqttd_sm:register_session(CleanSess, ClientId, sess_info(Session)),
{noreply, start_collector(Session), hibernate}; noreply(start_collector(Session));
handle_info({'EXIT', ClientPid, _Reason}, Session = #session{clean_sess = true, handle_info({'EXIT', ClientPid, _Reason}, Session = #session{clean_sess = true,
client_pid = ClientPid}) -> client_pid = ClientPid}) ->
{stop, normal, Session}; {stop, normal, Session};
handle_info({'EXIT', ClientPid, Reason}, Session = #session{clean_sess = false, handle_info({'EXIT', ClientPid, Reason}, Session = #session{clean_sess = false,
client_id = ClientId,
client_pid = ClientPid, client_pid = ClientPid,
expired_after = Expires}) -> expired_after = Expires}) ->
lager:info("Session(~s): unlink with client ~p: reason=~p", ?LOG(info, "Client ~p EXIT for ~p", [ClientPid, Reason], Session),
[ClientId, ClientPid, Reason]), TRef = timer(Expires, expired),
TRef = timer(Expires, session_expired), erlang:garbage_collect(), %%TODO: ???
noreply(Session#session{client_pid = undefined, expired_timer = TRef}); noreply(Session#session{client_pid = undefined, expired_timer = TRef});
handle_info({'EXIT', Pid, Reason}, Session = #session{client_id = ClientId, handle_info({'EXIT', Pid, Reason}, Session = #session{client_pid = ClientPid}) ->
client_pid = ClientPid}) ->
lager:error("Session(~s): Unexpected EXIT: client_pid=~p, exit_pid=~p, reason=~p", ?LOG(error, "Unexpected EXIT: client_pid=~p, exit_pid=~p, reason=~p",
[ClientId, ClientPid, Pid, Reason]), [ClientPid, Pid, Reason], Session),
noreply(Session); noreply(Session);
handle_info(session_expired, Session = #session{client_id = ClientId}) -> handle_info(expired, Session) ->
lager:error("Session(~s) expired, shutdown now.", [ClientId]), ?LOG(info, "expired, shutdown now.", [], Session),
{stop, {shutdown, expired}, Session}; {stop, {shutdown, expired}, Session};
handle_info(Info, Session = #session{client_id = ClientId}) -> handle_info(Info, Session) ->
lager:error("Session(~s) unexpected info: ~p", [ClientId, Info]), ?LOG(critical, "Unexpected info: ~p", [Info], Session),
{noreply, Session}. {noreply, Session}.
terminate(_Reason, #session{clean_sess = CleanSess, client_id = ClientId}) -> terminate(_Reason, #session{clean_sess = CleanSess, client_id = ClientId}) ->
@ -585,6 +576,17 @@ code_change(_OldVsn, Session, _Extra) ->
%%% Internal functions %%% Internal functions
%%%============================================================================= %%%=============================================================================
%%------------------------------------------------------------------------------
%% Kick old client out
%%------------------------------------------------------------------------------
kick(_ClientId, undefined, _Pid) ->
ignore;
kick(_ClientId, Pid, Pid) ->
ignore;
kick(ClientId, OldPid, Pid) ->
unlink(OldPid),
OldPid ! {shutdown, conflict, {ClientId, Pid}}.
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
%% Check inflight and awaiting_rel %% Check inflight and awaiting_rel
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
@ -658,7 +660,7 @@ acked(PktId, Session = #session{client_id = ClientId,
{_, Msg} -> {_, Msg} ->
emqttd_broker:foreach_hooks('message.acked', [ClientId, Msg]); emqttd_broker:foreach_hooks('message.acked', [ClientId, Msg]);
false -> false ->
lager:error("Session(~s): Cannot find acked message: ~p", [PktId]) ?LOG(error, "Cannot find acked pktid: ~p", [PktId], Session)
end, end,
Session#session{awaiting_ack = maps:remove(PktId, Awaiting), Session#session{awaiting_ack = maps:remove(PktId, Awaiting),
inflight_queue = lists:keydelete(PktId, 1, InflightQ)}. inflight_queue = lists:keydelete(PktId, 1, InflightQ)}.
@ -687,15 +689,15 @@ start_collector(Session = #session{collect_interval = Interval}) ->
TRef = erlang:send_after(timer:seconds(Interval), self(), collect_info), TRef = erlang:send_after(timer:seconds(Interval), self(), collect_info),
Session#session{collect_timer = TRef}. Session#session{collect_timer = TRef}.
info(#session{clean_sess = CleanSess, sess_info(#session{clean_sess = CleanSess,
subscriptions = Subscriptions, subscriptions = Subscriptions,
inflight_queue = InflightQueue, inflight_queue = InflightQueue,
max_inflight = MaxInflight, max_inflight = MaxInflight,
message_queue = MessageQueue, message_queue = MessageQueue,
awaiting_rel = AwaitingRel, awaiting_rel = AwaitingRel,
awaiting_ack = AwaitingAck, awaiting_ack = AwaitingAck,
awaiting_comp = AwaitingComp, awaiting_comp = AwaitingComp,
timestamp = CreatedAt}) -> timestamp = CreatedAt}) ->
Stats = emqttd_mqueue:stats(MessageQueue), Stats = emqttd_mqueue:stats(MessageQueue),
[{clean_sess, CleanSess}, [{clean_sess, CleanSess},
{subscriptions, Subscriptions}, {subscriptions, Subscriptions},

36
src/emqttd_sm.erl
View File

@ -57,7 +57,10 @@
-define(SM_POOL, ?MODULE). -define(SM_POOL, ?MODULE).
-define(SESSION_TIMEOUT, 60000). -define(CALL_TIMEOUT, 60000).
-define(LOG(Level, Format, Args, Session),
lager:Level("SM(~s): " ++ Format, [Session#mqtt_session.client_id | Args])).
%%%============================================================================= %%%=============================================================================
%%% Mnesia callbacks %%% Mnesia callbacks
@ -113,7 +116,7 @@ start_session(CleanSess, ClientId) ->
lookup_session(ClientId) -> lookup_session(ClientId) ->
case mnesia:dirty_read(session, ClientId) of case mnesia:dirty_read(session, ClientId) of
[Session] -> Session; [Session] -> Session;
[] -> undefined [] -> undefined
end. end.
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
@ -137,13 +140,11 @@ register_session(CleanSess, ClientId, Info) ->
unregister_session(CleanSess, ClientId) -> unregister_session(CleanSess, ClientId) ->
ets:delete(sesstab(CleanSess), {ClientId, self()}). ets:delete(sesstab(CleanSess), {ClientId, self()}).
sesstab(true) -> sesstab(true) -> mqtt_transient_session;
mqtt_transient_session; sesstab(false) -> mqtt_persistent_session.
sesstab(false) ->
mqtt_persistent_session.
call(SM, Req) -> call(SM, Req) ->
gen_server2:call(SM, Req, ?SESSION_TIMEOUT). %%infinity). gen_server2:call(SM, Req, ?CALL_TIMEOUT). %%infinity).
%%%============================================================================= %%%=============================================================================
%%% gen_server callbacks %%% gen_server callbacks
@ -223,8 +224,8 @@ create_session(CleanSess, ClientId, ClientPid) ->
case insert_session(Session) of case insert_session(Session) of
{aborted, {conflict, ConflictPid}} -> {aborted, {conflict, ConflictPid}} ->
%% Conflict with othe node? %% Conflict with othe node?
lager:error("Session(~s): Conflict with ~p!", [ClientId, ConflictPid]), lager:error("SM(~s): Conflict with ~p", [ClientId, ConflictPid]),
{error, conflict}; {error, mnesia_conflict};
{atomic, ok} -> {atomic, ok} ->
erlang:monitor(process, SessPid), erlang:monitor(process, SessPid),
{ok, SessPid} {ok, SessPid}
@ -245,8 +246,8 @@ insert_session(Session = #mqtt_session{client_id = ClientId}) ->
end). end).
%% Local node %% Local node
resume_session(#mqtt_session{client_id = ClientId, resume_session(Session = #mqtt_session{client_id = ClientId,
sess_pid = SessPid}, ClientPid) sess_pid = SessPid}, ClientPid)
when node(SessPid) =:= node() -> when node(SessPid) =:= node() ->
case is_process_alive(SessPid) of case is_process_alive(SessPid) of
@ -254,7 +255,7 @@ resume_session(#mqtt_session{client_id = ClientId,
emqttd_session:resume(SessPid, ClientId, ClientPid), emqttd_session:resume(SessPid, ClientId, ClientPid),
{ok, SessPid}; {ok, SessPid};
false -> false ->
lager:error("Session(~s): Cannot resume ~p, it seems already dead!", [ClientId, SessPid]), ?LOG(error, "Cannot resume ~p which seems already dead!", [SessPid], Session),
{error, session_died} {error, session_died}
end; end;
@ -265,12 +266,11 @@ resume_session(Session = #mqtt_session{client_id = ClientId, sess_pid = SessPid}
ok -> ok ->
{ok, SessPid}; {ok, SessPid};
{badrpc, nodedown} -> {badrpc, nodedown} ->
lager:error("Session(~s): Died for node ~s down!", [ClientId, Node]), ?LOG(error, "Session died for node '~s' down", [Node], Session),
remove_session(Session), remove_session(Session),
{error, session_nodedown}; {error, session_nodedown};
{badrpc, Reason} -> {badrpc, Reason} ->
lager:error("Session(~s): Failed to resume from node ~s for ~p", ?LOG(error, "Failed to resume from node ~s for ~p", [Node, Reason], Session),
[ClientId, Node, Reason]),
{error, Reason} {error, Reason}
end. end.
@ -288,11 +288,11 @@ destroy_session(Session = #mqtt_session{client_id = ClientId,
ok -> ok ->
remove_session(Session); remove_session(Session);
{badrpc, nodedown} -> {badrpc, nodedown} ->
lager:error("Session(~s): Died for node ~s down!", [ClientId, Node]), ?LOG(error, "Node '~s' down", [Node], Session),
remove_session(Session); remove_session(Session);
{badrpc, Reason} -> {badrpc, Reason} ->
lager:error("Session(~s): Failed to destory ~p on remote node ~p for ~s", ?LOG(error, "Failed to destory ~p on remote node ~p for ~s",
[ClientId, SessPid, Node, Reason]), [SessPid, Node, Reason], Session),
{error, Reason} {error, Reason}
end. end.