|
|
@@ -43,6 +43,8 @@
|
|
|
-include("emqx.hrl").
|
|
|
-include("emqx_mqtt.hrl").
|
|
|
|
|
|
+-include("logger.hrl").
|
|
|
+
|
|
|
-export([start_link/1]).
|
|
|
-export([info/1, attrs/1]).
|
|
|
-export([stats/1]).
|
|
|
@@ -155,8 +157,7 @@
|
|
|
|
|
|
-export_type([attr/0]).
|
|
|
|
|
|
--define(LOG(Level, Format, Args, _State),
|
|
|
- emqx_logger:Level("[Session] " ++ Format, Args)).
|
|
|
+-define(LOG(Level, Format, Args), ?LOG_LZ(Level, "[Session] " ++ Format, Args)).
|
|
|
|
|
|
%% @doc Start a session proc.
|
|
|
-spec(start_link(SessAttrs :: map()) -> {ok, pid()}).
|
|
|
@@ -390,11 +391,11 @@ handle_call(stats, _From, State) ->
|
|
|
reply(stats(State), State);
|
|
|
|
|
|
handle_call({discard, ByPid}, _From, State = #state{conn_pid = undefined}) ->
|
|
|
- ?LOG(warning, "Discarded by ~p", [ByPid], State),
|
|
|
+ ?LOG(warning, "Discarded by ~p", [ByPid]),
|
|
|
{stop, {shutdown, discarded}, ok, State};
|
|
|
|
|
|
handle_call({discard, ByPid}, _From, State = #state{client_id = ClientId, conn_pid = ConnPid}) ->
|
|
|
- ?LOG(warning, "Conn ~p is discarded by ~p", [ConnPid, ByPid], State),
|
|
|
+ ?LOG(warning, "Conn ~p is discarded by ~p", [ConnPid, ByPid]),
|
|
|
ConnPid ! {shutdown, discard, {ClientId, ByPid}},
|
|
|
{stop, {shutdown, discarded}, ok, State};
|
|
|
|
|
|
@@ -413,7 +414,7 @@ handle_call({register_publish_packet_id, PacketId, Ts}, _From,
|
|
|
{ok, ensure_stats_timer(ensure_await_rel_timer(State1))}
|
|
|
end;
|
|
|
true ->
|
|
|
- ?LOG(warning, "Dropped qos2 packet ~w for too many awaiting_rel", [PacketId], State),
|
|
|
+ ?LOG(warning, "Dropped qos2 packet ~w for too many awaiting_rel", [PacketId]),
|
|
|
emqx_metrics:trans(inc, 'messages/qos2/dropped'),
|
|
|
{{error, ?RC_RECEIVE_MAXIMUM_EXCEEDED}, State}
|
|
|
end);
|
|
|
@@ -425,7 +426,7 @@ handle_call({pubrec, PacketId, _ReasonCode}, _From, State = #state{inflight = In
|
|
|
true ->
|
|
|
{ok, ensure_stats_timer(acked(pubrec, PacketId, State))};
|
|
|
false ->
|
|
|
- ?LOG(warning, "The PUBREC PacketId ~w is not found.", [PacketId], State),
|
|
|
+ ?LOG(warning, "The PUBREC PacketId ~w is not found.", [PacketId]),
|
|
|
emqx_metrics:trans(inc, 'packets/pubrec/missed'),
|
|
|
{{error, ?RC_PACKET_IDENTIFIER_NOT_FOUND}, State}
|
|
|
end);
|
|
|
@@ -437,7 +438,7 @@ handle_call({pubrel, PacketId, _ReasonCode}, _From, State = #state{awaiting_rel
|
|
|
{_Ts, AwaitingRel1} ->
|
|
|
{ok, ensure_stats_timer(State#state{awaiting_rel = AwaitingRel1})};
|
|
|
error ->
|
|
|
- ?LOG(warning, "The PUBREL PacketId ~w is not found", [PacketId], State),
|
|
|
+ ?LOG(warning, "The PUBREL PacketId ~w is not found", [PacketId]),
|
|
|
emqx_metrics:trans(inc, 'packets/pubrel/missed'),
|
|
|
{{error, ?RC_PACKET_IDENTIFIER_NOT_FOUND}, State}
|
|
|
end);
|
|
|
@@ -496,7 +497,7 @@ handle_cast({puback, PacketId, _ReasonCode}, State = #state{inflight = Inflight}
|
|
|
true ->
|
|
|
ensure_stats_timer(dequeue(acked(puback, PacketId, State)));
|
|
|
false ->
|
|
|
- ?LOG(warning, "The PUBACK PacketId ~w is not found", [PacketId], State),
|
|
|
+ ?LOG(warning, "The PUBACK PacketId ~w is not found", [PacketId]),
|
|
|
emqx_metrics:trans(inc, 'packets/puback/missed'),
|
|
|
State
|
|
|
end);
|
|
|
@@ -508,7 +509,7 @@ handle_cast({pubcomp, PacketId, _ReasonCode}, State = #state{inflight = Inflight
|
|
|
true ->
|
|
|
ensure_stats_timer(dequeue(acked(pubcomp, PacketId, State)));
|
|
|
false ->
|
|
|
- ?LOG(warning, "The PUBCOMP PacketId ~w is not found", [PacketId], State),
|
|
|
+ ?LOG(warning, "The PUBCOMP PacketId ~w is not found", [PacketId]),
|
|
|
emqx_metrics:trans(inc, 'packets/pubcomp/missed'),
|
|
|
State
|
|
|
end);
|
|
|
@@ -526,14 +527,14 @@ handle_cast({resume, #{conn_pid := ConnPid,
|
|
|
expiry_timer = ExpireTimer,
|
|
|
will_delay_timer = WillDelayTimer}) ->
|
|
|
|
|
|
- ?LOG(info, "Resumed by connection ~p ", [ConnPid], State),
|
|
|
+ ?LOG(info, "Resumed by connection ~p ", [ConnPid]),
|
|
|
|
|
|
%% Cancel Timers
|
|
|
lists:foreach(fun emqx_misc:cancel_timer/1,
|
|
|
[RetryTimer, AwaitTimer, ExpireTimer, WillDelayTimer]),
|
|
|
|
|
|
case kick(ClientId, OldConnPid, ConnPid) of
|
|
|
- ok -> ?LOG(warning, "Connection ~p kickout ~p", [ConnPid, OldConnPid], State);
|
|
|
+ ok -> ?LOG(warning, "Connection ~p kickout ~p", [ConnPid, OldConnPid]);
|
|
|
ignore -> ok
|
|
|
end,
|
|
|
|
|
|
@@ -614,12 +615,12 @@ handle_info({timeout, Timer, emit_stats},
|
|
|
GcState1 = emqx_gc:reset(GcState),
|
|
|
{noreply, NewState#state{gc_state = GcState1}, hibernate};
|
|
|
{shutdown, Reason} ->
|
|
|
- ?LOG(warning, "shutdown due to ~p", [Reason], NewState),
|
|
|
+ ?LOG(warning, "shutdown due to ~p", [Reason]),
|
|
|
shutdown(Reason, NewState)
|
|
|
end;
|
|
|
|
|
|
handle_info({timeout, Timer, expired}, State = #state{expiry_timer = Timer}) ->
|
|
|
- ?LOG(info, "expired, shutdown now.", [], State),
|
|
|
+ ?LOG(info, "expired, shutdown now.", []),
|
|
|
shutdown(expired, State);
|
|
|
|
|
|
handle_info({timeout, Timer, will_delay}, State = #state{will_msg = WillMsg, will_delay_timer = Timer}) ->
|
|
|
@@ -645,7 +646,7 @@ handle_info({'EXIT', OldPid, _Reason}, State = #state{old_conn_pid = OldPid}) ->
|
|
|
|
|
|
handle_info({'EXIT', Pid, Reason}, State = #state{conn_pid = ConnPid}) ->
|
|
|
?LOG(error, "Unexpected EXIT: conn_pid=~p, exit_pid=~p, reason=~p",
|
|
|
- [ConnPid, Pid, Reason], State),
|
|
|
+ [ConnPid, Pid, Reason]),
|
|
|
{noreply, State};
|
|
|
|
|
|
handle_info(Info, State) ->
|
|
|
@@ -780,7 +781,7 @@ expire_awaiting_rel([{PacketId, Ts} | More], Now,
|
|
|
case (timer:now_diff(Now, Ts) div 1000) of
|
|
|
Age when Age >= Timeout ->
|
|
|
emqx_metrics:trans(inc, 'messages/qos2/expired'),
|
|
|
- ?LOG(warning, "Dropped qos2 packet ~s for await_rel_timeout", [PacketId], State),
|
|
|
+ ?LOG(warning, "Dropped qos2 packet ~s for await_rel_timeout", [PacketId]),
|
|
|
expire_awaiting_rel(More, Now, State#state{awaiting_rel = maps:remove(PacketId, AwaitingRel)});
|
|
|
Age ->
|
|
|
ensure_await_rel_timer(Timeout - max(0, Age), State)
|
|
|
@@ -895,7 +896,7 @@ acked(puback, PacketId, State = #state{client_id = ClientId, username = Username
|
|
|
emqx_hooks:run('message.acked', [#{client_id => ClientId, username => Username}], Msg),
|
|
|
State#state{inflight = emqx_inflight:delete(PacketId, Inflight)};
|
|
|
none ->
|
|
|
- ?LOG(warning, "Duplicated PUBACK PacketId ~w", [PacketId], State),
|
|
|
+ ?LOG(warning, "Duplicated PUBACK PacketId ~w", [PacketId]),
|
|
|
State
|
|
|
end;
|
|
|
|
|
|
@@ -905,10 +906,10 @@ acked(pubrec, PacketId, State = #state{client_id = ClientId, username = Username
|
|
|
emqx_hooks:run('message.acked', [#{client_id => ClientId, username => Username}], Msg),
|
|
|
State#state{inflight = emqx_inflight:update(PacketId, {pubrel, PacketId, os:timestamp()}, Inflight)};
|
|
|
{value, {pubrel, PacketId, _Ts}} ->
|
|
|
- ?LOG(warning, "Duplicated PUBREC PacketId ~w", [PacketId], State),
|
|
|
+ ?LOG(warning, "Duplicated PUBREC PacketId ~w", [PacketId]),
|
|
|
State;
|
|
|
none ->
|
|
|
- ?LOG(warning, "Unexpected PUBREC PacketId ~w", [PacketId], State),
|
|
|
+ ?LOG(warning, "Unexpected PUBREC PacketId ~w", [PacketId]),
|
|
|
State
|
|
|
end;
|
|
|
|