|
|
@@ -14,41 +14,43 @@
|
|
|
%% limitations under the License.
|
|
|
%%--------------------------------------------------------------------
|
|
|
|
|
|
-%% MQTT TCP/SSL Connection
|
|
|
+%% MQTT/TCP Connection
|
|
|
-module(emqx_connection).
|
|
|
|
|
|
--behaviour(gen_statem).
|
|
|
-
|
|
|
-include("emqx.hrl").
|
|
|
-include("emqx_mqtt.hrl").
|
|
|
-include("logger.hrl").
|
|
|
-include("types.hrl").
|
|
|
|
|
|
--logger_header("[Connection]").
|
|
|
+-logger_header("[MQTT]").
|
|
|
|
|
|
--export([start_link/3]).
|
|
|
+%% API
|
|
|
+-export([ start_link/3
|
|
|
+ , stop/1
|
|
|
+ ]).
|
|
|
|
|
|
-%% APIs
|
|
|
-export([ info/1
|
|
|
, stats/1
|
|
|
]).
|
|
|
|
|
|
-export([call/2]).
|
|
|
|
|
|
-%% state callbacks
|
|
|
--export([ idle/3
|
|
|
- , connected/3
|
|
|
- , disconnected/3
|
|
|
- ]).
|
|
|
+%% callback
|
|
|
+-export([init/4]).
|
|
|
|
|
|
-%% gen_statem callbacks
|
|
|
--export([ init/1
|
|
|
- , callback_mode/0
|
|
|
- , code_change/4
|
|
|
- , terminate/3
|
|
|
+%% Sys callbacks
|
|
|
+-export([ system_continue/3
|
|
|
+ , system_terminate/4
|
|
|
+ , system_code_change/4
|
|
|
+ , system_get_state/1
|
|
|
]).
|
|
|
|
|
|
+%% Internal callbacks
|
|
|
+-export([wakeup_from_hib/2]).
|
|
|
+
|
|
|
-record(state, {
|
|
|
+ %% Parent
|
|
|
+ parent :: pid(),
|
|
|
%% TCP/TLS Transport
|
|
|
transport :: esockd:transport(),
|
|
|
%% TCP/TLS Socket
|
|
|
@@ -57,10 +59,10 @@
|
|
|
peername :: emqx_types:peername(),
|
|
|
%% Sockname of the connection
|
|
|
sockname :: emqx_types:peername(),
|
|
|
+ %% Sock state
|
|
|
+ sockstate :: emqx_types:sockstate(),
|
|
|
%% The {active, N} option
|
|
|
active_n :: pos_integer(),
|
|
|
- %% The active state
|
|
|
- active_state :: running | blocked,
|
|
|
%% Publish Limit
|
|
|
pub_limit :: maybe(esockd_rate_limit:bucket()),
|
|
|
%% Rate Limit
|
|
|
@@ -72,94 +74,107 @@
|
|
|
%% Serialize function
|
|
|
serialize :: emqx_frame:serialize_fun(),
|
|
|
%% Channel State
|
|
|
- chan_state :: emqx_channel:channel()
|
|
|
+ channel :: emqx_channel:channel(),
|
|
|
+ %% Idle timer
|
|
|
+ idle_timer :: reference()
|
|
|
}).
|
|
|
|
|
|
-type(state() :: #state{}).
|
|
|
|
|
|
-define(ACTIVE_N, 100).
|
|
|
--define(HANDLE(T, C, D), handle((T), (C), (D))).
|
|
|
--define(INFO_KEYS, [socktype, peername, sockname, active_n, active_state,
|
|
|
+-define(INFO_KEYS, [socktype, peername, sockname, sockstate, active_n,
|
|
|
pub_limit, rate_limit]).
|
|
|
-define(CONN_STATS, [recv_pkt, recv_msg, send_pkt, send_msg]).
|
|
|
-define(SOCK_STATS, [recv_oct, recv_cnt, send_oct, send_cnt, send_pend]).
|
|
|
|
|
|
-%% @doc Start the connection.
|
|
|
-spec(start_link(esockd:transport(), esockd:socket(), proplists:proplist())
|
|
|
-> {ok, pid()}).
|
|
|
start_link(Transport, Socket, Options) ->
|
|
|
- {ok, proc_lib:spawn_link(?MODULE, init, [{Transport, Socket, Options}])}.
|
|
|
+ CPid = proc_lib:spawn_link(?MODULE, init, [self(), Transport, Socket, Options]),
|
|
|
+ {ok, CPid}.
|
|
|
|
|
|
%%--------------------------------------------------------------------
|
|
|
%% API
|
|
|
%%--------------------------------------------------------------------
|
|
|
|
|
|
-%% @doc Get infos of the connection.
|
|
|
+%% @doc Get infos of the connection/channel.
|
|
|
-spec(info(pid()|state()) -> emqx_types:infos()).
|
|
|
info(CPid) when is_pid(CPid) ->
|
|
|
call(CPid, info);
|
|
|
-info(Conn = #state{chan_state = ChanState}) ->
|
|
|
- ChanInfo = emqx_channel:info(ChanState),
|
|
|
- SockInfo = maps:from_list(info(?INFO_KEYS, Conn)),
|
|
|
+info(State = #state{channel = Channel}) ->
|
|
|
+ ChanInfo = emqx_channel:info(Channel),
|
|
|
+ SockInfo = maps:from_list(info(?INFO_KEYS, State)),
|
|
|
maps:merge(ChanInfo, #{sockinfo => SockInfo}).
|
|
|
|
|
|
-info(Keys, Conn) when is_list(Keys) ->
|
|
|
- [{Key, info(Key, Conn)} || Key <- Keys];
|
|
|
+info(Keys, State) when is_list(Keys) ->
|
|
|
+ [{Key, info(Key, State)} || Key <- Keys];
|
|
|
info(socktype, #state{transport = Transport, socket = Socket}) ->
|
|
|
Transport:type(Socket);
|
|
|
info(peername, #state{peername = Peername}) ->
|
|
|
Peername;
|
|
|
info(sockname, #state{sockname = Sockname}) ->
|
|
|
Sockname;
|
|
|
+info(sockstate, #state{sockstate = SockSt}) ->
|
|
|
+ SockSt;
|
|
|
info(active_n, #state{active_n = ActiveN}) ->
|
|
|
ActiveN;
|
|
|
-info(active_state, #state{active_state = ActiveSt}) ->
|
|
|
- ActiveSt;
|
|
|
info(pub_limit, #state{pub_limit = PubLimit}) ->
|
|
|
limit_info(PubLimit);
|
|
|
info(rate_limit, #state{rate_limit = RateLimit}) ->
|
|
|
limit_info(RateLimit);
|
|
|
-info(chan_state, #state{chan_state = ChanState}) ->
|
|
|
- emqx_channel:info(ChanState).
|
|
|
+info(channel, #state{channel = Channel}) ->
|
|
|
+ emqx_channel:info(Channel).
|
|
|
|
|
|
limit_info(Limit) ->
|
|
|
emqx_misc:maybe_apply(fun esockd_rate_limit:info/1, Limit).
|
|
|
|
|
|
-%% @doc Get stats of the channel.
|
|
|
+%% @doc Get stats of the connection/channel.
|
|
|
-spec(stats(pid()|state()) -> emqx_types:stats()).
|
|
|
stats(CPid) when is_pid(CPid) ->
|
|
|
call(CPid, stats);
|
|
|
-stats(#state{transport = Transport,
|
|
|
- socket = Socket,
|
|
|
- chan_state = ChanState}) ->
|
|
|
+stats(#state{transport = Transport,
|
|
|
+ socket = Socket,
|
|
|
+ channel = Channel}) ->
|
|
|
SockStats = case Transport:getstat(Socket, ?SOCK_STATS) of
|
|
|
{ok, Ss} -> Ss;
|
|
|
{error, _} -> []
|
|
|
end,
|
|
|
ConnStats = emqx_pd:get_counters(?CONN_STATS),
|
|
|
- ChanStats = emqx_channel:stats(ChanState),
|
|
|
+ ChanStats = emqx_channel:stats(Channel),
|
|
|
ProcStats = emqx_misc:proc_stats(),
|
|
|
lists:append([SockStats, ConnStats, ChanStats, ProcStats]).
|
|
|
|
|
|
-%% kick|discard|takeover
|
|
|
--spec(call(pid(), Req :: term()) -> Reply :: term()).
|
|
|
-call(CPid, Req) -> gen_statem:call(CPid, Req).
|
|
|
+call(Pid, Req) ->
|
|
|
+ gen_server:call(Pid, Req, infinity).
|
|
|
+
|
|
|
+stop(Pid) ->
|
|
|
+ gen_server:stop(Pid).
|
|
|
|
|
|
%%--------------------------------------------------------------------
|
|
|
-%% gen_statem callbacks
|
|
|
+%% callbacks
|
|
|
%%--------------------------------------------------------------------
|
|
|
|
|
|
-init({Transport, RawSocket, Options}) ->
|
|
|
+init(Parent, Transport, RawSocket, Options) ->
|
|
|
case Transport:wait(RawSocket) of
|
|
|
- {ok, Socket} ->
|
|
|
- do_init(Transport, Socket, Options);
|
|
|
- {error, Reason} ->
|
|
|
- ?LOG(warning, "connection failed to establish: ~p", [Reason])
|
|
|
+ {ok, Socket} ->
|
|
|
+ do_init(Parent, Transport, Socket, Options);
|
|
|
+ {error, Reason} when Reason =:= enotconn;
|
|
|
+ Reason =:= einval;
|
|
|
+ Reason =:= closed ->
|
|
|
+ Transport:fast_close(RawSocket),
|
|
|
+ exit(normal);
|
|
|
+ {error, timeout} ->
|
|
|
+ Transport:fast_close(RawSocket),
|
|
|
+ exit({shutdown, ssl_upgrade_timeout});
|
|
|
+ {error, Reason} ->
|
|
|
+ Transport:fast_close(RawSocket),
|
|
|
+ exit(Reason)
|
|
|
end.
|
|
|
|
|
|
-do_init(Transport, Socket, Options) ->
|
|
|
+do_init(Parent, Transport, Socket, Options) ->
|
|
|
{ok, Peername} = Transport:ensure_ok_or_exit(peername, [Socket]),
|
|
|
{ok, Sockname} = Transport:ensure_ok_or_exit(sockname, [Socket]),
|
|
|
+ emqx_logger:set_metadata_peername(esockd_net:format(Peername)),
|
|
|
Peercert = Transport:ensure_ok_or_exit(peercert, [Socket]),
|
|
|
ConnInfo = #{socktype => Transport:type(Socket),
|
|
|
peername => Peername,
|
|
|
@@ -167,7 +182,6 @@ do_init(Transport, Socket, Options) ->
|
|
|
peercert => Peercert,
|
|
|
conn_mod => ?MODULE
|
|
|
},
|
|
|
- emqx_logger:set_metadata_peername(esockd_net:format(Peername)),
|
|
|
Zone = proplists:get_value(zone, Options),
|
|
|
ActiveN = proplists:get_value(active_n, Options, ?ACTIVE_N),
|
|
|
PubLimit = init_limiter(emqx_zone:get_env(Zone, publish_limit)),
|
|
|
@@ -175,328 +189,315 @@ do_init(Transport, Socket, Options) ->
|
|
|
FrameOpts = emqx_zone:frame_options(Zone),
|
|
|
ParseState = emqx_frame:initial_parse_state(FrameOpts),
|
|
|
Serialize = emqx_frame:serialize_fun(),
|
|
|
- ChanState = emqx_channel:init(ConnInfo, Options),
|
|
|
- State = #state{transport = Transport,
|
|
|
- socket = Socket,
|
|
|
- peername = Peername,
|
|
|
- sockname = Sockname,
|
|
|
- active_n = ActiveN,
|
|
|
- active_state = running,
|
|
|
- pub_limit = PubLimit,
|
|
|
- rate_limit = RateLimit,
|
|
|
- parse_state = ParseState,
|
|
|
- serialize = Serialize,
|
|
|
- chan_state = ChanState
|
|
|
- },
|
|
|
+ Channel = emqx_channel:init(ConnInfo, Options),
|
|
|
IdleTimout = emqx_zone:get_env(Zone, idle_timeout, 30000),
|
|
|
- gen_statem:enter_loop(?MODULE, [{hibernate_after, 2 * IdleTimout}],
|
|
|
- idle, State, self(), [IdleTimout]).
|
|
|
+ IdleTimer = emqx_misc:start_timer(IdleTimout, idle_timeout),
|
|
|
+ HibAfterTimeout = emqx_zone:get_env(Zone, hibernate_after, IdleTimout*2),
|
|
|
+ State = #state{parent = Parent,
|
|
|
+ transport = Transport,
|
|
|
+ socket = Socket,
|
|
|
+ peername = Peername,
|
|
|
+ sockname = Sockname,
|
|
|
+ sockstate = idle,
|
|
|
+ active_n = ActiveN,
|
|
|
+ pub_limit = PubLimit,
|
|
|
+ rate_limit = RateLimit,
|
|
|
+ parse_state = ParseState,
|
|
|
+ serialize = Serialize,
|
|
|
+ channel = Channel,
|
|
|
+ idle_timer = IdleTimer
|
|
|
+ },
|
|
|
+ case activate_socket(State) of
|
|
|
+ {ok, NState} ->
|
|
|
+ recvloop(NState, #{hibernate_after => HibAfterTimeout});
|
|
|
+ {error, Reason} when Reason =:= einval;
|
|
|
+ Reason =:= enotconn;
|
|
|
+ Reason =:= closed ->
|
|
|
+ Transport:fast_close(Socket),
|
|
|
+ exit(normal);
|
|
|
+ {error, Reason} ->
|
|
|
+ Transport:fast_close(Socket),
|
|
|
+ erlang:exit({shutdown, Reason})
|
|
|
+ end.
|
|
|
|
|
|
-compile({inline, [init_limiter/1]}).
|
|
|
init_limiter(undefined) -> undefined;
|
|
|
init_limiter({Rate, Burst}) ->
|
|
|
esockd_rate_limit:new(Rate, Burst).
|
|
|
|
|
|
--compile({inline, [callback_mode/0]}).
|
|
|
-callback_mode() ->
|
|
|
- [state_functions, state_enter].
|
|
|
-
|
|
|
%%--------------------------------------------------------------------
|
|
|
-%% Idle State
|
|
|
-
|
|
|
-idle(enter, _, State) ->
|
|
|
- case activate_socket(State) of
|
|
|
- ok -> keep_state_and_data;
|
|
|
- {error, Reason} ->
|
|
|
- shutdown(Reason, State)
|
|
|
- end;
|
|
|
-
|
|
|
-idle(timeout, _Timeout, State) ->
|
|
|
- shutdown(idle_timeout, State);
|
|
|
-
|
|
|
-idle(cast, {incoming, Packet = ?CONNECT_PACKET(ConnPkt)}, State) ->
|
|
|
- SuccFun = fun(NewSt) -> {next_state, connected, NewSt} end,
|
|
|
- Serialize = emqx_frame:serialize_fun(ConnPkt),
|
|
|
- NState = State#state{serialize = Serialize},
|
|
|
- handle_incoming(Packet, SuccFun, NState);
|
|
|
-
|
|
|
-idle(cast, {incoming, Packet}, State) when is_record(Packet, mqtt_packet) ->
|
|
|
- SuccFun = fun(NewSt) -> {next_state, connected, NewSt} end,
|
|
|
- handle_incoming(Packet, SuccFun, State);
|
|
|
-
|
|
|
-idle(cast, {incoming, FrameError = {frame_error, _Reason}}, State) ->
|
|
|
- handle_incoming(FrameError, State);
|
|
|
-
|
|
|
-idle(EventType, Content, State) ->
|
|
|
- ?HANDLE(EventType, Content, State).
|
|
|
-
|
|
|
-%%--------------------------------------------------------------------
|
|
|
-%% Connected State
|
|
|
-
|
|
|
-connected(enter, _PrevSt, State) ->
|
|
|
- ok = register_self(State),
|
|
|
- keep_state_and_data;
|
|
|
-
|
|
|
-connected(cast, {incoming, Packet}, State) when is_record(Packet, mqtt_packet) ->
|
|
|
- handle_incoming(Packet, fun keep_state/1, State);
|
|
|
-
|
|
|
-connected(cast, {incoming, FrameError = {frame_error, _Reason}}, State) ->
|
|
|
- handle_incoming(FrameError, State);
|
|
|
-
|
|
|
-connected(info, Deliver = {deliver, _Topic, _Msg}, State) ->
|
|
|
- handle_deliver(emqx_misc:drain_deliver([Deliver]), State);
|
|
|
-
|
|
|
-connected(EventType, Content, State) ->
|
|
|
- ?HANDLE(EventType, Content, State).
|
|
|
-
|
|
|
-%%--------------------------------------------------------------------
|
|
|
-%% Disconnected State
|
|
|
-
|
|
|
-disconnected(enter, _, State = #state{chan_state = ChanState}) ->
|
|
|
- case emqx_channel:handle_info(disconnected, ChanState) of
|
|
|
- {ok, NChanState} ->
|
|
|
- ok = register_self(State#state{chan_state = NChanState}),
|
|
|
- keep_state(State#state{chan_state = NChanState});
|
|
|
- {stop, Reason, NChanState} ->
|
|
|
- stop(Reason, State#state{chan_state = NChanState})
|
|
|
- end;
|
|
|
+%% Recv Loop
|
|
|
+
|
|
|
+recvloop(State = #state{parent = Parent},
|
|
|
+ Options = #{hibernate_after := HibAfterTimeout}) ->
|
|
|
+ receive
|
|
|
+ {system, From, Request} ->
|
|
|
+ sys:handle_system_msg(Request, From, Parent,
|
|
|
+ ?MODULE, [], {State, Options});
|
|
|
+ {'EXIT', Parent, Reason} ->
|
|
|
+ terminate(Reason, State);
|
|
|
+ Msg ->
|
|
|
+ process_msg([Msg], State, Options)
|
|
|
+ after
|
|
|
+ HibAfterTimeout ->
|
|
|
+ hibernate(State, Options)
|
|
|
+ end.
|
|
|
|
|
|
-disconnected(info, Deliver = {deliver, _Topic, _Msg}, State) ->
|
|
|
- handle_deliver([Deliver], State);
|
|
|
+hibernate(State, Options) ->
|
|
|
+ proc_lib:hibernate(?MODULE, wakeup_from_hib, [State, Options]).
|
|
|
|
|
|
-disconnected(EventType, Content, State) ->
|
|
|
- ?HANDLE(EventType, Content, State).
|
|
|
+wakeup_from_hib(State, Options) ->
|
|
|
+ %% Maybe do something later here.
|
|
|
+ recvloop(State, Options).
|
|
|
|
|
|
%%--------------------------------------------------------------------
|
|
|
-%% Handle call
|
|
|
-
|
|
|
-handle({call, From}, info, State) ->
|
|
|
- reply(From, info(State), State);
|
|
|
-
|
|
|
-handle({call, From}, stats, State) ->
|
|
|
- reply(From, stats(State), State);
|
|
|
-
|
|
|
-handle({call, From}, state, State) ->
|
|
|
- reply(From, State, State);
|
|
|
-
|
|
|
-handle({call, From}, Req, State = #state{chan_state = ChanState}) ->
|
|
|
- case emqx_channel:handle_call(Req, ChanState) of
|
|
|
- {ok, Reply, NChanState} ->
|
|
|
- reply(From, Reply, State#state{chan_state = NChanState});
|
|
|
- {stop, Reason, Reply, NChanState} ->
|
|
|
- ok = gen_statem:reply(From, Reply),
|
|
|
- stop(Reason, State#state{chan_state = NChanState});
|
|
|
- {stop, Reason, Packet, Reply, NChanState} ->
|
|
|
- handle_outgoing(Packet, State#state{chan_state = NChanState}),
|
|
|
- ok = gen_statem:reply(From, Reply),
|
|
|
- stop(Reason, State#state{chan_state = NChanState})
|
|
|
- end;
|
|
|
+%% Process next Msg
|
|
|
+
|
|
|
+process_msg([], State, Options) ->
|
|
|
+ recvloop(State, Options);
|
|
|
+
|
|
|
+process_msg([Msg|More], State, Options) ->
|
|
|
+ case catch handle_msg(Msg, State) of
|
|
|
+ ok ->
|
|
|
+ process_msg(More, State, Options);
|
|
|
+ {ok, NState} ->
|
|
|
+ process_msg(More, NState, Options);
|
|
|
+ {ok, NextMsgs, NState} ->
|
|
|
+ process_msg(append_msg(NextMsgs, More), NState, Options);
|
|
|
+ {stop, Reason} ->
|
|
|
+ terminate(Reason, State);
|
|
|
+ {stop, Reason, NState} ->
|
|
|
+ terminate(Reason, NState);
|
|
|
+ {'EXIT', Reason} ->
|
|
|
+ terminate(Reason, State)
|
|
|
+ end.
|
|
|
|
|
|
%%--------------------------------------------------------------------
|
|
|
-%% Handle cast
|
|
|
-
|
|
|
-handle(cast, Msg, State = #state{chan_state = ChanState}) ->
|
|
|
- case emqx_channel:handle_info(Msg, ChanState) of
|
|
|
- ok -> {ok, State};
|
|
|
- {ok, NChanState} ->
|
|
|
- keep_state(State#state{chan_state = NChanState});
|
|
|
- {stop, Reason, NChanState} ->
|
|
|
- stop(Reason, State#state{chan_state = NChanState})
|
|
|
+%% Handle a Msg
|
|
|
+
|
|
|
+handle_msg({'$gen_call', From, Req}, State) ->
|
|
|
+ case handle_call(From, Req, State) of
|
|
|
+ {reply, Reply, NState} ->
|
|
|
+ gen_server:reply(From, Reply),
|
|
|
+ {ok, NState};
|
|
|
+ {stop, Reason, Reply, NState} ->
|
|
|
+ gen_server:reply(From, Reply),
|
|
|
+ stop(Reason, NState)
|
|
|
end;
|
|
|
|
|
|
-%%--------------------------------------------------------------------
|
|
|
-%% Handle info
|
|
|
-
|
|
|
-%% Handle incoming data
|
|
|
-handle(info, {Inet, _Sock, Data}, State = #state{chan_state = ChanState})
|
|
|
+handle_msg({Inet, _Sock, Data}, State = #state{channel = Channel})
|
|
|
when Inet == tcp; Inet == ssl ->
|
|
|
?LOG(debug, "RECV ~p", [Data]),
|
|
|
Oct = iolist_size(Data),
|
|
|
emqx_pd:update_counter(incoming_bytes, Oct),
|
|
|
ok = emqx_metrics:inc('bytes.received', Oct),
|
|
|
- NChanState = emqx_channel:received(Oct, ChanState),
|
|
|
- NState = State#state{chan_state = NChanState},
|
|
|
- process_incoming(Data, NState);
|
|
|
+ {ok, NChannel} = emqx_channel:handle_in(Oct, Channel),
|
|
|
+ process_incoming(Data, State#state{channel = NChannel});
|
|
|
+
|
|
|
+handle_msg({incoming, Packet = ?CONNECT_PACKET(ConnPkt)},
|
|
|
+ State = #state{idle_timer = IdleTimer}) ->
|
|
|
+ ok = emqx_misc:cancel_timer(IdleTimer),
|
|
|
+ Serialize = emqx_frame:serialize_fun(ConnPkt),
|
|
|
+ NState = State#state{serialize = Serialize,
|
|
|
+ idle_timer = undefined
|
|
|
+ },
|
|
|
+ handle_incoming(Packet, NState);
|
|
|
|
|
|
-handle(info, {Error, _Sock, Reason}, State)
|
|
|
+handle_msg({incoming, Packet}, State) ->
|
|
|
+ handle_incoming(Packet, State);
|
|
|
+
|
|
|
+handle_msg({Error, _Sock, Reason}, State)
|
|
|
when Error == tcp_error; Error == ssl_error ->
|
|
|
- shutdown(Reason, State);
|
|
|
+ handle_info({sock_error, Reason}, State);
|
|
|
|
|
|
-handle(info, {Closed, _Sock}, State)
|
|
|
+handle_msg({Closed, _Sock}, State)
|
|
|
when Closed == tcp_closed; Closed == ssl_closed ->
|
|
|
- {next_state, disconnected, State};
|
|
|
+ handle_info(sock_closed, State);
|
|
|
|
|
|
-handle(info, {Passive, _Sock}, State)
|
|
|
+handle_msg({Passive, _Sock}, State)
|
|
|
when Passive == tcp_passive; Passive == ssl_passive ->
|
|
|
- %% Rate limit here:)
|
|
|
+ %% Rate limit and activate socket here.
|
|
|
NState = ensure_rate_limit(State),
|
|
|
case activate_socket(NState) of
|
|
|
- ok -> keep_state(NState);
|
|
|
+ {ok, NState} -> {ok, NState};
|
|
|
{error, Reason} ->
|
|
|
- shutdown(Reason, NState)
|
|
|
+ {ok, {sock_error, Reason}, NState}
|
|
|
end;
|
|
|
|
|
|
-handle(info, activate_socket, State) ->
|
|
|
- %% Rate limit timer expired.
|
|
|
- NState = State#state{active_state = running,
|
|
|
- limit_timer = undefined
|
|
|
+%% Rate limit timer expired.
|
|
|
+handle_msg(activate_socket, State) ->
|
|
|
+ NState = State#state{sockstate = idle,
|
|
|
+ limit_timer = undefined
|
|
|
},
|
|
|
case activate_socket(NState) of
|
|
|
- ok -> keep_state(NState);
|
|
|
+ {ok, NState} -> {ok, NState};
|
|
|
{error, Reason} ->
|
|
|
- shutdown(Reason, NState)
|
|
|
+ {ok, {sock_error, Reason}, State}
|
|
|
end;
|
|
|
|
|
|
-handle(info, {inet_reply, _Sock, ok}, _State) ->
|
|
|
- %% something sent
|
|
|
- keep_state_and_data;
|
|
|
+handle_msg(Deliver = {deliver, _Topic, _Msg},
|
|
|
+ State = #state{channel = Channel}) ->
|
|
|
+ Delivers = emqx_misc:drain_deliver([Deliver]),
|
|
|
+ Result = emqx_channel:handle_out(Delivers, Channel),
|
|
|
+ handle_return(Result, State);
|
|
|
|
|
|
-handle(info, {inet_reply, _Sock, {error, Reason}}, State) ->
|
|
|
- shutdown(Reason, State);
|
|
|
+handle_msg({outgoing, Packets}, State) ->
|
|
|
+ {ok, handle_outgoing(Packets, State)};
|
|
|
|
|
|
-handle(info, {timeout, TRef, keepalive},
|
|
|
- State = #state{transport = Transport, socket = Socket}) ->
|
|
|
- case Transport:getstat(Socket, [recv_oct]) of
|
|
|
- {ok, [{recv_oct, RecvOct}]} ->
|
|
|
- handle_timeout(TRef, {keepalive, RecvOct}, State);
|
|
|
- {error, Reason} ->
|
|
|
- shutdown(Reason, State)
|
|
|
- end;
|
|
|
+%% something sent
|
|
|
+handle_msg({inet_reply, _Sock, ok}, _State) ->
|
|
|
+ ok;
|
|
|
|
|
|
-handle(info, {timeout, TRef, emit_stats}, State) ->
|
|
|
- handle_timeout(TRef, {emit_stats, stats(State)}, State);
|
|
|
+handle_msg({inet_reply, _Sock, {error, Reason}}, State) ->
|
|
|
+ handle_info({sock_error, Reason}, State);
|
|
|
|
|
|
-handle(info, {timeout, TRef, Msg}, State) ->
|
|
|
- handle_timeout(TRef, Msg, State);
|
|
|
+handle_msg({timeout, TRef, TMsg}, State) ->
|
|
|
+ handle_timeout(TRef, TMsg, State);
|
|
|
|
|
|
-handle(info, {shutdown, Reason}, State) ->
|
|
|
- shutdown(Reason, State);
|
|
|
+handle_msg(Shutdown = {shutdown, _Reason}, State) ->
|
|
|
+ stop(Shutdown, State);
|
|
|
|
|
|
-handle(info, Info, State = #state{chan_state = ChanState}) ->
|
|
|
- case emqx_channel:handle_info(Info, ChanState) of
|
|
|
- {ok, NChanState} ->
|
|
|
- keep_state(State#state{chan_state = NChanState});
|
|
|
- {stop, Reason, NChanState} ->
|
|
|
- stop(Reason, State#state{chan_state = NChanState})
|
|
|
- end.
|
|
|
+handle_msg(Msg, State) -> handle_info(Msg, State).
|
|
|
|
|
|
-code_change(_Vsn, State, Data, _Extra) ->
|
|
|
- {ok, State, Data}.
|
|
|
+%%--------------------------------------------------------------------
|
|
|
+%% Terminate
|
|
|
|
|
|
-terminate(Reason, _StateName, #state{transport = Transport,
|
|
|
- socket = Socket,
|
|
|
- chan_state = ChanState
|
|
|
- }) ->
|
|
|
+terminate(Reason, #state{transport = Transport,
|
|
|
+ socket = Socket,
|
|
|
+ sockstate = SockSt,
|
|
|
+ channel = Channel}) ->
|
|
|
?LOG(debug, "Terminated for ~p", [Reason]),
|
|
|
- ok = Transport:fast_close(Socket),
|
|
|
- emqx_channel:terminate(Reason, ChanState).
|
|
|
+ SockSt =:= closed orelse Transport:fast_close(Socket),
|
|
|
+ emqx_channel:terminate(Reason, Channel),
|
|
|
+ exit(Reason).
|
|
|
+
|
|
|
+%%--------------------------------------------------------------------
|
|
|
+%% Sys callbacks
|
|
|
+
|
|
|
+system_continue(_Parent, _Deb, {State, Options}) ->
|
|
|
+ recvloop(State, Options).
|
|
|
+
|
|
|
+system_terminate(Reason, _Parent, _Deb, {State, _}) ->
|
|
|
+ terminate(Reason, State).
|
|
|
+
|
|
|
+system_code_change(Misc, _, _, _) ->
|
|
|
+ {ok, Misc}.
|
|
|
+
|
|
|
+system_get_state({State, _Options}) ->
|
|
|
+ {ok, State}.
|
|
|
|
|
|
%%--------------------------------------------------------------------
|
|
|
-%% Internal functions
|
|
|
+%% Handle call
|
|
|
+
|
|
|
+handle_call(_From, info, State) ->
|
|
|
+ {reply, info(State), State};
|
|
|
+
|
|
|
+handle_call(_From, stats, State) ->
|
|
|
+ {reply, stats(State), State};
|
|
|
+
|
|
|
+handle_call(_From, Req, State = #state{channel = Channel}) ->
|
|
|
+ case emqx_channel:handle_call(Req, Channel) of
|
|
|
+ {reply, Reply, NChannel} ->
|
|
|
+ {reply, Reply, State#state{channel = NChannel}};
|
|
|
+ {stop, Reason, Reply, NChannel} ->
|
|
|
+ {stop, Reason, Reply, State#state{channel = NChannel}};
|
|
|
+ {stop, Reason, Reply, OutPacket, NChannel} ->
|
|
|
+ NState = State#state{channel = NChannel},
|
|
|
+ NState1 = handle_outgoing(OutPacket, NState),
|
|
|
+ {stop, Reason, Reply, NState1}
|
|
|
+ end.
|
|
|
+
|
|
|
%%--------------------------------------------------------------------
|
|
|
+%% Handle timeout
|
|
|
|
|
|
-register_self(State = #state{active_n = ActiveN,
|
|
|
- active_state = ActiveSt,
|
|
|
- chan_state = ChanState
|
|
|
- }) ->
|
|
|
- ChanAttrs = emqx_channel:attrs(ChanState),
|
|
|
- SockAttrs = #{active_n => ActiveN,
|
|
|
- active_state => ActiveSt
|
|
|
- },
|
|
|
- Attrs = maps:merge(ChanAttrs, #{sockinfo => SockAttrs}),
|
|
|
- emqx_channel:handle_info({register, Attrs, stats(State)}, ChanState).
|
|
|
+handle_timeout(TRef, idle_timeout, State = #state{idle_timer = TRef}) ->
|
|
|
+ stop(idle_timeout, State);
|
|
|
+
|
|
|
+handle_timeout(TRef, emit_stats, State) ->
|
|
|
+ handle_timeout(TRef, {emit_stats, stats(State)}, State);
|
|
|
+
|
|
|
+handle_timeout(TRef, keepalive, State = #state{transport = Transport,
|
|
|
+ socket = Socket}) ->
|
|
|
+ case Transport:getstat(Socket, [recv_oct]) of
|
|
|
+ {ok, [{recv_oct, RecvOct}]} ->
|
|
|
+ handle_timeout(TRef, {keepalive, RecvOct}, State);
|
|
|
+ {error, Reason} ->
|
|
|
+ handle_info({sockerr, Reason}, State)
|
|
|
+ end;
|
|
|
+
|
|
|
+handle_timeout(TRef, Msg, State = #state{channel = Channel}) ->
|
|
|
+ handle_return(emqx_channel:handle_timeout(TRef, Msg, Channel), State).
|
|
|
|
|
|
%%--------------------------------------------------------------------
|
|
|
-%% Process incoming data
|
|
|
+%% Process/Parse incoming data.
|
|
|
|
|
|
-compile({inline, [process_incoming/2]}).
|
|
|
process_incoming(Data, State) ->
|
|
|
- process_incoming(Data, [], State).
|
|
|
+ {Packets, NState} = parse_incoming(Data, State),
|
|
|
+ {ok, next_incoming_msgs(Packets), NState}.
|
|
|
+
|
|
|
+-compile({inline, [parse_incoming/2]}).
|
|
|
+parse_incoming(Data, State) ->
|
|
|
+ parse_incoming(Data, [], State).
|
|
|
|
|
|
-process_incoming(<<>>, Packets, State) ->
|
|
|
- keep_state(State, next_incoming_events(Packets));
|
|
|
+parse_incoming(<<>>, Packets, State) ->
|
|
|
+ {Packets, State};
|
|
|
|
|
|
-process_incoming(Data, Packets, State = #state{parse_state = ParseState}) ->
|
|
|
+parse_incoming(Data, Packets, State = #state{parse_state = ParseState}) ->
|
|
|
try emqx_frame:parse(Data, ParseState) of
|
|
|
{more, NParseState} ->
|
|
|
- NState = State#state{parse_state = NParseState},
|
|
|
- keep_state(NState, next_incoming_events(Packets));
|
|
|
+ {Packets, State#state{parse_state = NParseState}};
|
|
|
{ok, Packet, Rest, NParseState} ->
|
|
|
NState = State#state{parse_state = NParseState},
|
|
|
- process_incoming(Rest, [Packet|Packets], NState)
|
|
|
+ parse_incoming(Rest, [Packet|Packets], NState)
|
|
|
catch
|
|
|
error:Reason:Stk ->
|
|
|
?LOG(error, "~nParse failed for ~p~nStacktrace: ~p~nFrame data:~p",
|
|
|
[Reason, Stk, Data]),
|
|
|
- keep_state(State, next_incoming_events(Packets++[{frame_error, Reason}]))
|
|
|
+ {[{frame_error, Reason}|Packets], State}
|
|
|
end.
|
|
|
|
|
|
--compile({inline, [next_incoming_events/1]}).
|
|
|
-next_incoming_events([]) -> [];
|
|
|
-next_incoming_events(Packets) ->
|
|
|
- [next_event(cast, {incoming, Packet}) || Packet <- Packets].
|
|
|
+next_incoming_msgs([Packet]) ->
|
|
|
+ {incoming, Packet};
|
|
|
+next_incoming_msgs(Packets) ->
|
|
|
+ [{incoming, Packet} || Packet <- lists:reverse(Packets)].
|
|
|
|
|
|
%%--------------------------------------------------------------------
|
|
|
%% Handle incoming packet
|
|
|
|
|
|
-handle_incoming(Packet = ?PACKET(Type), SuccFun, State = #state{chan_state = ChanState}) ->
|
|
|
+handle_incoming(Packet = ?PACKET(Type), State = #state{channel = Channel}) ->
|
|
|
_ = inc_incoming_stats(Type),
|
|
|
- _ = emqx_metrics:inc_recv(Packet),
|
|
|
+ ok = emqx_metrics:inc_recv(Packet),
|
|
|
?LOG(debug, "RECV ~s", [emqx_packet:format(Packet)]),
|
|
|
- case emqx_channel:handle_in(Packet, ChanState) of
|
|
|
- {ok, NChanState} ->
|
|
|
- SuccFun(State#state{chan_state= NChanState});
|
|
|
- {ok, OutPackets, NChanState} ->
|
|
|
- NState = State#state{chan_state = NChanState},
|
|
|
- handle_outgoing(OutPackets, SuccFun, NState);
|
|
|
- {close, Reason, NChanState} ->
|
|
|
- close(Reason, State#state{chan_state = NChanState});
|
|
|
- {close, Reason, OutPackets, NChanState} ->
|
|
|
- NState = State#state{chan_state= NChanState},
|
|
|
- close(Reason, handle_outgoing(OutPackets, fun(NewSt) -> NewSt end, NState));
|
|
|
- {stop, Reason, NChanState} ->
|
|
|
- stop(Reason, State#state{chan_state = NChanState});
|
|
|
- {stop, Reason, OutPackets, NChanState} ->
|
|
|
- NState = State#state{chan_state= NChanState},
|
|
|
- stop(Reason, handle_outgoing(OutPackets, fun(NewSt) -> NewSt end, NState))
|
|
|
- end.
|
|
|
-
|
|
|
-handle_incoming(FrameError = {frame_error, _Reason}, State = #state{chan_state = ChanState}) ->
|
|
|
- case emqx_channel:handle_in(FrameError, ChanState) of
|
|
|
- {close, Reason, NChanState} ->
|
|
|
- close(Reason, State#state{chan_state = NChanState});
|
|
|
- {close, Reason, OutPackets, NChanState} ->
|
|
|
- NState = State#state{chan_state= NChanState},
|
|
|
- close(Reason, handle_outgoing(OutPackets, fun(NewSt) -> NewSt end, NState));
|
|
|
- {stop, Reason, NChanState} ->
|
|
|
- stop(Reason, State#state{chan_state = NChanState});
|
|
|
- {stop, Reason, OutPackets, NChanState} ->
|
|
|
- NState = State#state{chan_state= NChanState},
|
|
|
- stop(Reason, handle_outgoing(OutPackets, fun(NewSt) -> NewSt end, NState))
|
|
|
- end.
|
|
|
+ handle_return(emqx_channel:handle_in(Packet, Channel), State);
|
|
|
|
|
|
-%%-------------------------------------------------------------------
|
|
|
-%% Handle deliver
|
|
|
+handle_incoming(FrameError, State = #state{channel = Channel}) ->
|
|
|
+ handle_return(emqx_channel:handle_in(FrameError, Channel), State).
|
|
|
|
|
|
-handle_deliver(Delivers, State = #state{chan_state = ChanState}) ->
|
|
|
- case emqx_channel:handle_out({deliver, Delivers}, ChanState) of
|
|
|
- {ok, NChanState} ->
|
|
|
- keep_state(State#state{chan_state = NChanState});
|
|
|
- {ok, Packets, NChanState} ->
|
|
|
- handle_outgoing(Packets, fun keep_state/1, State#state{chan_state = NChanState})
|
|
|
- end.
|
|
|
+%%--------------------------------------------------------------------
|
|
|
+%% Handle channel return
|
|
|
+
|
|
|
+handle_return(ok, State) ->
|
|
|
+ {ok, State};
|
|
|
+handle_return({ok, NChannel}, State) ->
|
|
|
+ {ok, State#state{channel = NChannel}};
|
|
|
+handle_return({ok, Replies, NChannel}, State) ->
|
|
|
+ {ok, next_msgs(Replies), State#state{channel = NChannel}};
|
|
|
+handle_return({stop, Reason, NChannel}, State) ->
|
|
|
+ stop(Reason, State#state{channel = NChannel});
|
|
|
+handle_return({stop, Reason, OutPacket, NChannel}, State) ->
|
|
|
+ NState = State#state{channel = NChannel},
|
|
|
+ NState1 = handle_outgoing(OutPacket, NState),
|
|
|
+ stop(Reason, NState1).
|
|
|
|
|
|
%%--------------------------------------------------------------------
|
|
|
%% Handle outgoing packets
|
|
|
|
|
|
-handle_outgoing(Packet, State) ->
|
|
|
- handle_outgoing(Packet, fun (_) -> ok end, State).
|
|
|
-
|
|
|
-handle_outgoing(Packets, SuccFun, State) when is_list(Packets) ->
|
|
|
- send(lists:map(serialize_and_inc_stats_fun(State), Packets), SuccFun, State);
|
|
|
+handle_outgoing(Packets, State) when is_list(Packets) ->
|
|
|
+ send(lists:map(serialize_and_inc_stats_fun(State), Packets), State);
|
|
|
|
|
|
-handle_outgoing(Packet, SuccFun, State) ->
|
|
|
- send((serialize_and_inc_stats_fun(State))(Packet), SuccFun, State).
|
|
|
+handle_outgoing(Packet, State) ->
|
|
|
+ send((serialize_and_inc_stats_fun(State))(Packet), State).
|
|
|
|
|
|
serialize_and_inc_stats_fun(#state{serialize = Serialize}) ->
|
|
|
fun(Packet = ?PACKET(Type)) ->
|
|
|
@@ -514,39 +515,73 @@ serialize_and_inc_stats_fun(#state{serialize = Serialize}) ->
|
|
|
%%--------------------------------------------------------------------
|
|
|
%% Send data
|
|
|
|
|
|
-send(IoData, SuccFun, State = #state{transport = Transport,
|
|
|
- socket = Socket,
|
|
|
- chan_state = ChanState}) ->
|
|
|
+send(IoData, State = #state{transport = Transport,
|
|
|
+ socket = Socket,
|
|
|
+ channel = Channel}) ->
|
|
|
Oct = iolist_size(IoData),
|
|
|
ok = emqx_metrics:inc('bytes.sent', Oct),
|
|
|
case Transport:async_send(Socket, IoData) of
|
|
|
- ok -> NChanState = emqx_channel:sent(Oct, ChanState),
|
|
|
- SuccFun(State#state{chan_state = NChanState});
|
|
|
- {error, Reason} ->
|
|
|
- shutdown(Reason, State)
|
|
|
+ ok ->
|
|
|
+ {ok, NChannel} = emqx_channel:handle_out(Oct, Channel),
|
|
|
+ State#state{channel = NChannel};
|
|
|
+ Error = {error, _Reason} ->
|
|
|
+ %% Simulate an inet_reply to postpone handling the error
|
|
|
+ self() ! {inet_reply, Socket, Error}, State
|
|
|
end.
|
|
|
|
|
|
%%--------------------------------------------------------------------
|
|
|
-%% Handle timeout
|
|
|
+%% Handle Info
|
|
|
+
|
|
|
+handle_info({enter, _}, State = #state{active_n = ActiveN,
|
|
|
+ sockstate = SockSt,
|
|
|
+ channel = Channel}) ->
|
|
|
+ ChanAttrs = emqx_channel:attrs(Channel),
|
|
|
+ SockAttrs = #{active_n => ActiveN,
|
|
|
+ sockstate => SockSt
|
|
|
+ },
|
|
|
+ Attrs = maps:merge(ChanAttrs, #{sockinfo => SockAttrs}),
|
|
|
+ handle_info({register, Attrs, stats(State)}, State);
|
|
|
+
|
|
|
+handle_info({sockerr, _Reason}, #state{sockstate = closed}) -> ok;
|
|
|
+handle_info({sockerr, Reason}, State) ->
|
|
|
+ ?LOG(debug, "Socket error: ~p", [Reason]),
|
|
|
+ handle_info({sock_closed, Reason}, close_socket(State));
|
|
|
+
|
|
|
+handle_info(sock_closed, #state{sockstate = closed}) -> ok;
|
|
|
+handle_info(sock_closed, State) ->
|
|
|
+ ?LOG(debug, "Socket closed"),
|
|
|
+ handle_info({sock_closed, closed}, close_socket(State));
|
|
|
|
|
|
-handle_timeout(TRef, Msg, State = #state{chan_state = ChanState}) ->
|
|
|
- case emqx_channel:handle_timeout(TRef, Msg, ChanState) of
|
|
|
- {ok, NChanState} ->
|
|
|
- keep_state(State#state{chan_state = NChanState});
|
|
|
- {ok, Packets, NChanState} ->
|
|
|
- handle_outgoing(Packets, fun keep_state/1, State#state{chan_state = NChanState});
|
|
|
- {close, Reason, NChanState} ->
|
|
|
- close(Reason, State#state{chan_state = NChanState});
|
|
|
- {close, Reason, OutPackets, NChanState} ->
|
|
|
- NState = State#state{chan_state= NChanState},
|
|
|
- close(Reason, handle_outgoing(OutPackets, fun(NewSt) -> NewSt end, NState));
|
|
|
- {stop, Reason, NChanState} ->
|
|
|
- stop(Reason, State#state{chan_state = NChanState});
|
|
|
- {stop, Reason, OutPackets, NChanState} ->
|
|
|
- NState = State#state{chan_state= NChanState},
|
|
|
- stop(Reason, handle_outgoing(OutPackets, fun(NewSt) -> NewSt end, NState))
|
|
|
+handle_info({close, Reason}, State) ->
|
|
|
+ ?LOG(debug, "Force close due to : ~p", [Reason]),
|
|
|
+ {ok, close_socket(State)};
|
|
|
+
|
|
|
+handle_info(Info, State = #state{channel = Channel}) ->
|
|
|
+ handle_return(emqx_channel:handle_info(Info, Channel), State).
|
|
|
+
|
|
|
+%%--------------------------------------------------------------------
|
|
|
+%% Activate Socket
|
|
|
+
|
|
|
+-compile({inline, [activate_socket/1]}).
|
|
|
+activate_socket(State = #state{sockstate = closed}) ->
|
|
|
+ {ok, State};
|
|
|
+activate_socket(State = #state{sockstate = blocked}) ->
|
|
|
+ {ok, State};
|
|
|
+activate_socket(State = #state{transport = Transport,
|
|
|
+ socket = Socket,
|
|
|
+ active_n = N}) ->
|
|
|
+ case Transport:setopts(Socket, [{active, N}]) of
|
|
|
+ ok -> {ok, State#state{sockstate = running}};
|
|
|
+ Error -> Error
|
|
|
end.
|
|
|
|
|
|
+%%--------------------------------------------------------------------
|
|
|
+%% Close Socket
|
|
|
+
|
|
|
+close_socket(State = #state{transport = Transport, socket = Socket}) ->
|
|
|
+ ok = Transport:fast_close(Socket),
|
|
|
+ State#state{sockstate = closed}.
|
|
|
+
|
|
|
%%--------------------------------------------------------------------
|
|
|
%% Ensure rate limit
|
|
|
|
|
|
@@ -568,22 +603,10 @@ ensure_rate_limit([{Rl, Pos, Cnt}|Limiters], State) ->
|
|
|
{Pause, Rl1} ->
|
|
|
?LOG(debug, "Pause ~pms due to rate limit", [Pause]),
|
|
|
TRef = erlang:send_after(Pause, self(), activate_socket),
|
|
|
- NState = State#state{active_state = blocked,
|
|
|
- limit_timer = TRef
|
|
|
- },
|
|
|
+ NState = State#state{sockstate = blocked, limit_timer = TRef},
|
|
|
setelement(Pos, NState, Rl1)
|
|
|
end.
|
|
|
|
|
|
-%%--------------------------------------------------------------------
|
|
|
-%% Activate Socket
|
|
|
-
|
|
|
--compile({inline, [activate_socket/1]}).
|
|
|
-activate_socket(#state{active_state = blocked}) -> ok;
|
|
|
-activate_socket(#state{transport = Transport,
|
|
|
- socket = Socket,
|
|
|
- active_n = N}) ->
|
|
|
- Transport:setopts(Socket, [{active, N}]).
|
|
|
-
|
|
|
%%--------------------------------------------------------------------
|
|
|
%% Inc incoming/outgoing stats
|
|
|
|
|
|
@@ -600,41 +623,25 @@ inc_incoming_stats(Type) when is_integer(Type) ->
|
|
|
-compile({inline, [inc_outgoing_stats/1]}).
|
|
|
inc_outgoing_stats(Type) ->
|
|
|
emqx_pd:update_counter(send_pkt, 1),
|
|
|
- (Type == ?PUBLISH)
|
|
|
- andalso emqx_pd:update_counter(send_msg, 1).
|
|
|
+ (Type == ?PUBLISH) andalso emqx_pd:update_counter(send_msg, 1).
|
|
|
|
|
|
%%--------------------------------------------------------------------
|
|
|
%% Helper functions
|
|
|
|
|
|
--compile({inline,
|
|
|
- [ reply/3
|
|
|
- , keep_state/1
|
|
|
- , keep_state/2
|
|
|
- , next_event/2
|
|
|
- , shutdown/2
|
|
|
- , stop/2
|
|
|
- ]}).
|
|
|
-
|
|
|
-reply(From, Reply, State) ->
|
|
|
- {keep_state, State, [{reply, From, Reply}]}.
|
|
|
-
|
|
|
-keep_state(State) ->
|
|
|
- {keep_state, State}.
|
|
|
-
|
|
|
-keep_state(State, Events) ->
|
|
|
- {keep_state, State, Events}.
|
|
|
-
|
|
|
-next_event(Type, Content) ->
|
|
|
- {next_event, Type, Content}.
|
|
|
-
|
|
|
-close(Reason, State = #state{transport = Transport, socket = Socket}) ->
|
|
|
- ?LOG(warning, "Closed for ~p", [Reason]),
|
|
|
- ok = Transport:fast_close(Socket),
|
|
|
- {next_state, disconnected, State}.
|
|
|
+-compile({inline, [append_msg/2]}).
|
|
|
+append_msg(Msgs, Q) when is_list(Msgs) ->
|
|
|
+ lists:append(Msgs, Q);
|
|
|
+append_msg(Msg, Q) -> [Msg|Q].
|
|
|
|
|
|
-shutdown(Reason, State) ->
|
|
|
- stop({shutdown, Reason}, State).
|
|
|
+-compile({inline, [next_msgs/1]}).
|
|
|
+next_msgs(Packet) when is_record(Packet, mqtt_packet) ->
|
|
|
+ {outgoing, Packet};
|
|
|
+next_msgs(Action) when is_tuple(Action) ->
|
|
|
+ Action;
|
|
|
+next_msgs(Actions) when is_list(Actions) ->
|
|
|
+ Actions.
|
|
|
|
|
|
+-compile({inline, [stop/2]}).
|
|
|
stop(Reason, State) ->
|
|
|
{stop, Reason, State}.
|
|
|
|
Feng Lee