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

Log macros for lazy evaluation

This commit is contained in:
terry-xiaoyu 2018-11-15 23:12:36 +08:00 committed by turtleDeng
parent 2269967f1a
commit 41315bff0d
4 changed files with 42 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
include/logger.hrl Normal file
View File

@ -0,0 +1,17 @@
%%--------------------------------------------------------------------
%% Logs with header
%%--------------------------------------------------------------------
-ifdef(LOG_HEADER).
%% with header
-define(LOG(Level, Format, Args),
begin
(emqx_logger:Level(#{},#{report_cb =>
fun(_) ->
{?LOG_HEADER ++ " "++ (Format), (Args)}
end}))
end).
-else.
%% without header
-define(LOG(Level, Format, Args),
emqx_logger:Level(Format, Args)).
-endif.

14
src/emqx_connection.erl
View File

@ -16,8 +16,11 @@
-behaviour(gen_server).
-define(LOG_HEADER, "[TCP]").
-include("emqx.hrl").
-include("emqx_mqtt.hrl").
-include("logger.hrl").
-export([start_link/3]).
-export([info/1, attrs/1]).
@ -50,13 +53,6 @@
-define(SOCK_STATS, [recv_oct, recv_cnt, send_oct, send_cnt, send_pend]).
-define(LOG(Level, Format, Args),
emqx_logger:Level(#{header => "[TCP] ", format => Format, args => Args},
#{report_cb =>
fun(#{header := Hdr0, format := Fmt0, args := Args0}) ->
{Hdr0 ++ Fmt0, Args0}
end})).
start_link(Transport, Socket, Options) ->
{ok, proc_lib:spawn_link(?MODULE, init, [[Transport, Socket, Options]])}.
@ -135,7 +131,7 @@ init([Transport, RawSocket, Options]) ->
PubLimit = init_limiter(emqx_zone:get_env(Zone, publish_limit)),
EnableStats = emqx_zone:get_env(Zone, enable_stats, true),
IdleTimout = emqx_zone:get_env(Zone, idle_timeout, 30000),
SendFun = send_fun(Transport, Socket, Peername),
SendFun = send_fun(Transport, Socket),
ProtoState = emqx_protocol:init(#{peername => Peername,
sockname => Sockname,
peercert => Peercert,
@ -169,7 +165,7 @@ init_limiter(undefined) ->
init_limiter({Rate, Burst}) ->
esockd_rate_limit:new(Rate, Burst).
send_fun(Transport, Socket, Peername) ->
send_fun(Transport, Socket) ->
fun(Packet, Options) ->
Data = emqx_frame:serialize(Packet, Options),
try Transport:async_send(Socket, Data) of

22
src/emqx_protocol.erl
View File

@ -14,8 +14,11 @@
-module(emqx_protocol).
-define(LOG_HEADER, "[MQTT]").
-include("emqx.hrl").
-include("emqx_mqtt.hrl").
-include("logger.hrl").
-export([init/2]).
-export([info/1]).
@ -72,13 +75,6 @@
-define(NO_PROPS, undefined).
-define(LOG(Level, Format, Args),
emqx_logger:Level(#{header => "[MQTT] ", format => Format, args => Args},
#{report_cb =>
fun(#{header := Hdr0, format := Fmt0, args := Args0}) ->
{Hdr0 ++ Fmt0, Args0}
end})).
%%------------------------------------------------------------------------------
%% Init
%%------------------------------------------------------------------------------
@ -810,17 +806,9 @@ check_sub_acl(TopicFilters, PState) ->
end, {ok, []}, TopicFilters).
trace(recv, Packet) ->
emqx_logger:debug(#{header => "[MQTT] RECV ~s", pck => Packet},
#{report_cb =>
fun(#{header := Fmt, pck := Pckt}) ->
{Fmt, [emqx_packet:format(Pckt)]}
end});
?LOG(debug, "RECV ~s", [emqx_packet:format(Packet)]);
trace(send, Packet) ->
emqx_logger:debug(#{header => "[MQTT] SEND ~s", pck => Packet},
#{report_cb =>
fun(#{header := Fmt, pck := Pckt}) ->
{Fmt, [emqx_packet:format(Pckt)]}
end}).
?LOG(debug, "SEND ~s", [emqx_packet:format(Packet)]).
inc_stats(recv, Type, PState = #pstate{recv_stats = Stats}) ->
PState#pstate{recv_stats = inc_stats(Type, Stats)};

33
src/emqx_ws_connection.erl
View File

@ -14,8 +14,11 @@
-module(emqx_ws_connection).
-define(LOG_HEADER, "[WS]").
-include("emqx.hrl").
-include("emqx_mqtt.hrl").
-include("logger.hrl").
-export([info/1, attrs/1]).
-export([stats/1]).
@ -45,12 +48,6 @@
-define(SOCK_STATS, [recv_oct, recv_cnt, send_oct, send_cnt]).
-define(WSLOG(Level, Format, Args),
emqx_logger:Level(#{header => "[WS] ", format => Format, args => Args},
#{report_cb =>
fun(#{header := Hdr0, format := Fmt0, args := Args0}) ->
{Hdr0 ++ Fmt0, Args0}
end})).
%%------------------------------------------------------------------------------
%% API
%%------------------------------------------------------------------------------
@ -169,7 +166,7 @@ websocket_handle({binary, Data}, State = #state{parser_state = ParserState,
proto_state = ProtoState}) ->
BinSize = iolist_size(Data),
put(recv_oct, get(recv_oct) + BinSize),
?WSLOG(debug, "RECV ~p", [Data]),
?LOG(debug, "RECV ~p", [Data]),
emqx_metrics:inc('bytes/received', BinSize),
case catch emqx_frame:parse(iolist_to_binary(Data), ParserState) of
{more, NewParserState} ->
@ -181,7 +178,7 @@ websocket_handle({binary, Data}, State = #state{parser_state = ParserState,
{ok, ProtoState1} ->
websocket_handle({binary, Rest}, reset_parser(State#state{proto_state = ProtoState1}));
{error, Error} ->
?WSLOG(error, "Protocol error - ~p", [Error]),
?LOG(error, "Protocol error - ~p", [Error]),
stop(Error, State);
{error, Reason, ProtoState1} ->
shutdown(Reason, State#state{proto_state = ProtoState1});
@ -189,10 +186,10 @@ websocket_handle({binary, Data}, State = #state{parser_state = ParserState,
stop(Error, State#state{proto_state = ProtoState1})
end;
{error, Error} ->
?WSLOG(error, "Frame error: ~p", [Error]),
?LOG(error, "Frame error: ~p", [Error]),
stop(Error, State);
{'EXIT', Reason} ->
?WSLOG(error, "Frame error:~p~nFrame data: ~p", [Reason, Data]),
?LOG(error, "Frame error:~p~nFrame data: ~p", [Reason, Data]),
shutdown(parse_error, State)
end.
@ -230,12 +227,12 @@ websocket_info({timeout, Timer, emit_stats},
{ok, State#state{stats_timer = undefined}, hibernate};
websocket_info({keepalive, start, Interval}, State) ->
?WSLOG(debug, "Keepalive at the interval of ~p", [Interval]),
?LOG(debug, "Keepalive at the interval of ~p", [Interval]),
case emqx_keepalive:start(stat_fun(), Interval, {keepalive, check}) of
{ok, KeepAlive} ->
{ok, State#state{keepalive = KeepAlive}};
{error, Error} ->
?WSLOG(warning, "Keepalive error - ~p", [Error]),
?LOG(warning, "Keepalive error - ~p", [Error]),
shutdown(Error, State)
end;
@ -244,19 +241,19 @@ websocket_info({keepalive, check}, State = #state{keepalive = KeepAlive}) ->
{ok, KeepAlive1} ->
{ok, State#state{keepalive = KeepAlive1}};
{error, timeout} ->
?WSLOG(debug, "Keepalive Timeout!", []),
?LOG(debug, "Keepalive Timeout!", []),
shutdown(keepalive_timeout, State);
{error, Error} ->
?WSLOG(warning, "Keepalive error - ~p", [Error]),
?LOG(warning, "Keepalive error - ~p", [Error]),
shutdown(keepalive_error, State)
end;
websocket_info({shutdown, discard, {ClientId, ByPid}}, State) ->
?WSLOG(warning, "discarded by ~s:~p", [ClientId, ByPid]),
?LOG(warning, "discarded by ~s:~p", [ClientId, ByPid]),
shutdown(discard, State);
websocket_info({shutdown, conflict, {ClientId, NewPid}}, State) ->
?WSLOG(warning, "clientid '~s' conflict with ~p", [ClientId, NewPid]),
?LOG(warning, "clientid '~s' conflict with ~p", [ClientId, NewPid]),
shutdown(conflict, State);
websocket_info({binary, Data}, State) ->
@ -266,13 +263,13 @@ websocket_info({shutdown, Reason}, State) ->
shutdown(Reason, State);
websocket_info(Info, State) ->
?WSLOG(error, "unexpected info: ~p", [Info]),
?LOG(error, "unexpected info: ~p", [Info]),
{ok, State}.
terminate(SockError, _Req, #state{keepalive = Keepalive,
proto_state = ProtoState,
shutdown = Shutdown}) ->
?WSLOG(debug, "Terminated for ~p, sockerror: ~p",
?LOG(debug, "Terminated for ~p, sockerror: ~p",
[Shutdown, SockError]),
emqx_keepalive:cancel(Keepalive),
case {ProtoState, Shutdown} of