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

refactor client

This commit is contained in:
Feng 2015-10-30 16:00:11 +08:00
parent 8d9270a6ba
commit f4cf90ae91
2 changed files with 161 additions and 157 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
rebar.config
View File

@ -29,8 +29,8 @@
{deps, [ {deps, [
{gproc, ".*", {git, "git://github.com/uwiger/gproc.git", {branch, "master"}}}, {gproc, ".*", {git, "git://github.com/uwiger/gproc.git", {branch, "master"}}},
{lager, ".*", {git, "git://github.com/basho/lager.git", {branch, "master"}}}, {lager, ".*", {git, "git://github.com/basho/lager.git", {branch, "master"}}},
{esockd, "2.*", {git, "git://github.com/emqtt/esockd.git", {branch, "master"}}}, {esockd, "3.*", {git, "git://github.com/emqtt/esockd.git", {branch, "3.0"}}},
{mochiweb, ".*", {git, "git://github.com/emqtt/mochiweb.git", {branch, "master"}}} {mochiweb, ".*", {git, "git://github.com/emqtt/mochiweb.git", {branch, "4.0"}}}
]}. ]}.
{recursive_cmds, [ct, eunit, clean]}. {recursive_cmds, [ct, eunit, clean]}.

312
src/emqttd_client.erl
View File

@ -24,7 +24,6 @@
%%% %%%
%%% @end %%% @end
%%%----------------------------------------------------------------------------- %%%-----------------------------------------------------------------------------
-module(emqttd_client). -module(emqttd_client).
-author("Feng Lee <feng@emqtt.io>"). -author("Feng Lee <feng@emqtt.io>").
@ -33,40 +32,35 @@
-include("emqttd_protocol.hrl"). -include("emqttd_protocol.hrl").
-include("emqttd_internel.hrl").
-behaviour(gen_server).
%% API Function Exports %% API Function Exports
-export([start_link/2, session/1, info/1, kick/1]). -export([start_link/2, session/1, info/1, kick/1]).
%% SUB/UNSUB Asynchronously %% SUB/UNSUB Asynchronously, called by plugins.
-export([subscribe/2, unsubscribe/2]). -export([subscribe/2, unsubscribe/2]).
-behaviour(gen_server).
%% 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,
code_change/3, terminate/2]). code_change/3, terminate/2]).
%% Client State... %% Client State
-record(state, {transport, -record(client_state, {connection, peername, peerhost, peerport,
socket, await_recv, conn_state, rate_limit,
peername, parser_fun, proto_state, packet_opts,
conn_name, keepalive}).
await_recv,
conn_state,
rate_limiter,
parser,
proto_state,
packet_opts,
keepalive}).
-define(DEBUG(Format, Args, State), -define(INFO_KEYS, [peername, peerhost, peerport, await_recv, conn_state]).
lager:debug("Client(~s): " ++ Format,
[emqttd_net:format(State#state.peername) | Args])).
-define(ERROR(Format, Args, State),
lager:error("Client(~s): " ++ Format,
[emqttd_net:format(State#state.peername) | Args])).
start_link(SockArgs, MqttEnv) -> -define(SOCK_STATS, [recv_oct, recv_cnt, send_oct, send_cnt]).
{ok, proc_lib:spawn_link(?MODULE, init, [[SockArgs, MqttEnv]])}.
-define(LOG(Level, Format, Args, State),
lager:Level("Client(~s): " ++ Format, [State#client_state.peername | Args])).
start_link(Connection, MqttEnv) ->
{ok, proc_lib:spawn_link(?MODULE, init, [[Connection, MqttEnv]])}.
session(CPid) -> session(CPid) ->
gen_server:call(CPid, session, infinity). gen_server:call(CPid, session, infinity).
@ -83,140 +77,156 @@ subscribe(CPid, TopicTable) ->
unsubscribe(CPid, Topics) -> unsubscribe(CPid, Topics) ->
gen_server:cast(CPid, {unsubscribe, Topics}). gen_server:cast(CPid, {unsubscribe, Topics}).
init([SockArgs = {Transport, Sock, _SockFun}, MqttEnv]) -> init([Connection0, MqttEnv]) ->
% Transform if ssl. {ok, Connection} = Connection0:wait(),
{ok, NewSock} = esockd_connection:accept(SockArgs), {PeerHost, PeerPort, PeerName} =
%%TODO:... case Connection:peername() of
{ok, BufSizes} = inet:getopts(Sock, [sndbuf, recbuf, buffer]), {ok, {Host, Port}} ->
io:format("~p~n", [BufSizes]), {Host, Port, esockd_net:format({Host, Port})};
{ok, Peername} = emqttd_net:peername(Sock), {error, enotconn} ->
{ok, ConnStr} = emqttd_net:connection_string(Sock, inbound), Connection:fast_close(),
SendFun = send_fun(Transport, NewSock), exit(normal);
PktOpts = proplists:get_value(packet, MqttEnv), {error, Reason} ->
ProtoState = emqttd_protocol:init(Peername, SendFun, PktOpts), Connection:fast_close(),
Limiter = proplists:get_value(rate_limiter, MqttEnv), exit({shutdown, Reason})
State = run_socket(#state{transport = Transport, end,
socket = NewSock, SendFun = fun(Data) ->
peername = Peername, try Connection:async_send(Data) of
conn_name = ConnStr, true -> ok
await_recv = false, catch
conn_state = running, error:Error -> exit({shutdown, Error})
rate_limiter = Limiter, end
packet_opts = PktOpts, end,
parser = emqttd_parser:new(PktOpts), PktOpts = proplists:get_value(packet, MqttEnv),
proto_state = ProtoState}), ParserFun = emqttd_parser:new(PktOpts),
ProtoState = emqttd_protocol:init(PeerName, SendFun, PktOpts),
RateLimit = proplists:get_value(rate_limit, Connection:opts()),
State = run_socket(#client_state{connection = Connection,
peername = PeerName,
peerhost = PeerHost,
peerport = PeerPort,
await_recv = false,
conn_state = running,
rate_limit = RateLimit,
parser_fun = ParserFun,
proto_state = ProtoState,
packet_opts = PktOpts}),
ClientOpts = proplists:get_value(client, MqttEnv), ClientOpts = proplists:get_value(client, MqttEnv),
IdleTimout = proplists:get_value(idle_timeout, ClientOpts, 10), IdleTimout = proplists:get_value(idle_timeout, ClientOpts, 10),
gen_server:enter_loop(?MODULE, [], State, timer:seconds(IdleTimout)). gen_server:enter_loop(?MODULE, [], State, timer:seconds(IdleTimout)).
handle_call(session, _From, State = #state{proto_state = ProtoState}) -> handle_call(session, _From, State = #client_state{proto_state = ProtoState}) ->
{reply, emqttd_protocol:session(ProtoState), State}; {reply, emqttd_protocol:session(ProtoState), State};
handle_call(info, _From, State = #state{conn_name = ConnName, handle_call(info, _From, State = #client_state{connection = Connection,
proto_state = ProtoState}) -> proto_state = ProtoState}) ->
{reply, [{conn_name, ConnName} | emqttd_protocol:info(ProtoState)], State}; ClientInfo = [{Key, Val} || {Key, Val} <- ?record_to_proplist(client_state, State), lists:member(Key, ?INFO_KEYS)],
ProtoInfo = emqttd_protocol:info(ProtoState),
{ok, SockStats} = Connection:getstat(?SOCK_STATS),
Info = lists:append([ClientInfo, [{proto_info, ProtoInfo}, {sock_stats, SockStats}]]),
{reply, Info, State};
handle_call(kick, _From, State) -> handle_call(kick, _From, State) ->
{stop, {shutdown, kick}, ok, State}; {stop, {shutdown, kick}, ok, State};
handle_call(Req, _From, State) -> handle_call(Req, _From, State) ->
?ERROR("Unexpected request: ~p", [Req], State), ?LOG(critical, "Unexpected request: ~p", [Req], State),
{reply, {error, unsupported_request}, State}. {reply, {error, unsupported_request}, State}.
handle_cast({subscribe, TopicTable}, State) -> handle_cast({subscribe, TopicTable}, State) ->
with_session(fun(SessPid) -> emqttd_session:subscribe(SessPid, TopicTable) end, State); with_session(fun(SessPid) ->
emqttd_session:subscribe(SessPid, TopicTable)
end, State);
handle_cast({unsubscribe, Topics}, State) -> handle_cast({unsubscribe, Topics}, State) ->
with_session(fun(SessPid) -> emqttd_session:unsubscribe(SessPid, Topics) end, State); with_session(fun(SessPid) ->
emqttd_session:unsubscribe(SessPid, Topics)
end, State);
handle_cast(Msg, State) -> handle_cast(Msg, State) ->
?ERROR("Unexpected msg: ~p",[Msg], State), ?LOG(critical, "Unexpected msg: ~p", [Msg], State),
{noreply, State}. noreply(State).
handle_info(timeout, State) -> handle_info(timeout, State) ->
stop({shutdown, timeout}, State); shutdown(idle_timeout, State);
%% Asynchronous SUBACK %% Asynchronous SUBACK
handle_info({suback, PacketId, GrantedQos}, State = #state{proto_state = ProtoState}) -> handle_info({suback, PacketId, GrantedQos}, State) ->
{ok, ProtoState1} = emqttd_protocol:send(?SUBACK_PACKET(PacketId, GrantedQos), ProtoState), with_proto_state(fun(ProtoState) ->
noreply(State#state{proto_state = ProtoState1}); Packet = ?SUBACK_PACKET(PacketId, GrantedQos),
emqttd_protocol:send(Packet, ProtoState)
end, State);
handle_info({deliver, Message}, State = #state{proto_state = ProtoState}) -> handle_info({deliver, Message}, State) ->
{ok, ProtoState1} = emqttd_protocol:send(Message, ProtoState), with_proto_state(fun(ProtoState) ->
noreply(State#state{proto_state = ProtoState1}); emqttd_protocol:send(Message, ProtoState)
end, State);
handle_info({redeliver, {?PUBREL, PacketId}}, State = #state{proto_state = ProtoState}) -> handle_info({redeliver, {?PUBREL, PacketId}}, State) ->
{ok, ProtoState1} = emqttd_protocol:redeliver({?PUBREL, PacketId}, ProtoState), with_proto_state(fun(ProtoState) ->
noreply(State#state{proto_state = ProtoState1}); emqttd_protocol:redeliver({?PUBREL, PacketId}, ProtoState)
end, State);
handle_info({stop, duplicate_id, _NewPid}, State=#state{proto_state = ProtoState, handle_info({shutdown, conflict, {ClientId, NewPid}}, State) ->
conn_name = ConnName}) -> ?LOG(warning, "clientid '~s' conflict with ~p", [ClientId, NewPid], State),
lager:warning("Shutdown for duplicate clientid: ~s, conn:~s", shutdown(confict, State);
[emqttd_protocol:clientid(ProtoState), ConnName]),
stop({shutdown, duplicate_id}, State);
handle_info(activate_sock, State) -> handle_info(activate_sock, State) ->
noreply(run_socket(State#state{conn_state = running})); noreply(run_socket(State#client_state{conn_state = running}));
handle_info({inet_async, Sock, _Ref, {ok, Data}}, State = #state{peername = Peername, socket = Sock}) -> handle_info({inet_async, _Sock, _Ref, {ok, Data}}, State) ->
Size = size(Data), Size = size(Data),
lager:debug("RECV from ~s: ~p", [emqttd_net:format(Peername), Data]), ?LOG(debug, "RECV: ~p", [Data], State),
emqttd_metrics:inc('bytes/received', Size), emqttd_metrics:inc('bytes/received', Size),
received(Data, rate_limit(Size, State#state{await_recv = false})); received(Data, rate_limit(Size, State#client_state{await_recv = false}));
handle_info({inet_async, _Sock, _Ref, {error, Reason}}, State) -> handle_info({inet_async, _Sock, _Ref, {error, Reason}}, State) ->
%%TODO: ... shutdown(Reason, State);
network_error(Reason, State);
handle_info({inet_reply, _Ref, ok}, State) -> handle_info({inet_reply, _Sock, ok}, State) ->
%%TODO: ok...
io:format("inet_reply ok~n"),
noreply(State); noreply(State);
handle_info({inet_reply, _Sock, {error, Reason}}, State) -> handle_info({inet_reply, _Sock, {error, Reason}}, State) ->
network_error(Reason, State); shutdown(Reason, State);
handle_info({keepalive, start, TimeoutSec}, State = #state{transport = Transport, socket = Socket}) -> handle_info({keepalive, start, Interval}, State = #client_state{connection = Connection}) ->
?DEBUG("Start KeepAlive with ~p seconds", [TimeoutSec], State), ?LOG(debug, "Keepalive at the interval of ~p", [Interval], State),
StatFun = fun() -> StatFun = fun() ->
case Transport:getstat(Socket, [recv_oct]) of case Connection:getstat([recv_oct]) of
{ok, [{recv_oct, RecvOct}]} -> {ok, RecvOct}; {ok, [{recv_oct, RecvOct}]} -> {ok, RecvOct};
{error, Error} -> {error, Error} {error, Error} -> {error, Error}
end end
end, end,
KeepAlive = emqttd_keepalive:start(StatFun, TimeoutSec, {keepalive, check}), KeepAlive = emqttd_keepalive:start(StatFun, Interval, {keepalive, check}),
noreply(State#state{keepalive = KeepAlive}); noreply(State#client_state{keepalive = KeepAlive});
handle_info({keepalive, check}, State = #state{keepalive = KeepAlive}) -> handle_info({keepalive, check}, State = #client_state{keepalive = KeepAlive}) ->
case emqttd_keepalive:check(KeepAlive) of case emqttd_keepalive:check(KeepAlive) of
{ok, KeepAlive1} -> {ok, KeepAlive1} ->
noreply(State#state{keepalive = KeepAlive1}); noreply(State#state{keepalive = KeepAlive1});
{error, timeout} -> {error, timeout} ->
?DEBUG("Keepalive Timeout!", [], State), ?LOG(debug, "Keepalive timeout", [], State),
stop({shutdown, keepalive_timeout}, State#state{keepalive = undefined}); shutdown(keepalive_timeout, State);
{error, Error} -> {error, Error} ->
?DEBUG("Keepalive Error - ~p", [Error], State), ?LOG(warning, "Keepalive error - ~p", [Error], State),
stop({shutdown, keepalive_error}, State#state{keepalive = undefined}) shutdown(Error, State)
end; end;
handle_info(Info, State) -> handle_info(Info, State) ->
?ERROR("Unexpected info: ~p", [Info], State), ?LOG(critical, "Unexpected info: ~p", [Info], State),
{noreply, State}. noreply(State).
terminate(Reason, #state{transport = Transport, terminate(Reason, #client_state{connection = Connection,
socket = Socket, keepalive = KeepAlive,
keepalive = KeepAlive, proto_state = ProtoState}) ->
proto_state = ProtoState}) -> Connection:fast_close(),
emqttd_keepalive:cancel(KeepAlive), emqttd_keepalive:cancel(KeepAlive),
if
Reason == {shutdown, conn_closed} -> ok;
true -> Transport:fast_close(Socket)
end,
case {ProtoState, Reason} of case {ProtoState, Reason} of
{undefined, _} -> ok; {undefined, _} ->
ok;
{_, {shutdown, Error}} -> {_, {shutdown, Error}} ->
emqttd_protocol:shutdown(Error, ProtoState); emqttd_protocol:shutdown(Error, ProtoState);
{_, Reason} -> {_, Reason} ->
emqttd_protocol:shutdown(Reason, ProtoState) emqttd_protocol:shutdown(Reason, ProtoState)
end. end.
@ -227,79 +237,73 @@ code_change(_OldVsn, State, _Extra) ->
%%% Internal functions %%% Internal functions
%%%============================================================================= %%%=============================================================================
noreply(State) -> with_proto_state(Fun, State = #client_state{proto_state = ProtoState}) ->
{noreply, State, hibernate}. {ok, ProtoState1} = Fun(ProtoState),
noreply(State#client_state{proto_state = ProtoState1}).
stop(Reason, State) -> with_session(Fun, State = #client_state{proto_state = ProtoState}) ->
{stop, Reason, State}. Fun(emqttd_protocol:session(ProtoState)),
noreply(State).
with_session(Fun, State = #state{proto_state = ProtoState}) ->
Fun(emqttd_protocol:session(ProtoState)), noreply(State).
%% receive and parse tcp data %% receive and parse tcp data
received(<<>>, State) -> received(<<>>, State) ->
noreply(State); noreply(State);
received(Bytes, State = #state{parser = Parser, received(Bytes, State = #client_state{parser_fun = ParserFun,
packet_opts = PacketOpts, packet_opts = PacketOpts,
proto_state = ProtoState}) -> proto_state = ProtoState}) ->
case catch Parser(Bytes) of case catch ParserFun(Bytes) of
{more, NewParser} -> {more, NewParser} ->
noreply(run_socket(State#state{parser = NewParser})); noreply(run_socket(State#client_state{parser_fun = NewParser}));
{ok, Packet, Rest} -> {ok, Packet, Rest} ->
emqttd_metrics:received(Packet), emqttd_metrics:received(Packet),
case emqttd_protocol:received(Packet, ProtoState) of case emqttd_protocol:received(Packet, ProtoState) of
{ok, ProtoState1} -> {ok, ProtoState1} ->
received(Rest, State#state{parser = emqttd_parser:new(PacketOpts), received(Rest, State#client_state{parser_fun = emqttd_parser:new(PacketOpts),
proto_state = ProtoState1}); proto_state = ProtoState1});
{error, Error} -> {error, Error} ->
?ERROR("Protocol error - ~p", [Error], State), ?LOG(error, "Protocol error - ~p", [Error], State),
stop({shutdown, Error}, State); shutdown(Error, State);
{error, Error, ProtoState1} -> {error, Error, ProtoState1} ->
stop({shutdown, Error}, State#state{proto_state = ProtoState1}); shutdown(Error, State#client_state{proto_state = ProtoState1});
{stop, Reason, ProtoState1} -> {stop, Reason, ProtoState1} ->
stop(Reason, State#state{proto_state = ProtoState1}) stop(Reason, State#client_state{proto_state = ProtoState1})
end; end;
{error, Error} -> {error, Error} ->
?ERROR("Framing error - ~p", [Error], State), ?LOG(error, "Framing error - ~p", [Error], State),
stop({shutdown, Error}, State); shutdown(Error, State);
{'EXIT', Reason} -> {'EXIT', Reason} ->
?ERROR("Parser failed for ~p~nError Frame: ~p", [Reason, Bytes], State), ?LOG(error, "Parser failed for ~p", [Reason], State),
{stop, {shutdown, frame_error}, State} ?LOG(error, "Error data: ~p", [Bytes], State),
shutdown(parser_error, State)
end. end.
network_error(Reason, State = #state{peername = Peername}) -> rate_limit(_Size, State = #client_state{rate_limit = undefined}) ->
lager:warning("Client(~s): network error - ~p",
[emqttd_net:format(Peername), Reason]),
stop({shutdown, conn_closed}, State).
rate_limit(_Size, State = #state{rate_limiter = undefined}) ->
run_socket(State); run_socket(State);
rate_limit(Size, State = #state{socket = Sock, rate_limiter = Limiter}) -> rate_limit(Size, State = #client_state{rate_limit = Limiter}) ->
{ok, BufSizes} = inet:getopts(Sock, [sndbuf, recbuf, buffer]), case esockd_ratelimit:check(Limiter, Size) of
io:format("~p~n", [BufSizes]),
case esockd_rate_limiter:check(Limiter, Size) of
{0, Limiter1} -> {0, Limiter1} ->
run_socket(State#state{conn_state = running, rate_limiter = Limiter1}); run_socket(State#client_state{conn_state = running, rate_limit = Limiter1});
{Pause, Limiter1} -> {Pause, Limiter1} ->
?ERROR("~p Received, Rate Limiter Pause for ~w", [Size, Pause], State), ?LOG(error, "Rate limiter pause for ~p", [Size, Pause], State),
erlang:send_after(Pause, self(), activate_sock), erlang:send_after(Pause, self(), activate_sock),
State#state{conn_state = blocked, rate_limiter = Limiter1} State#client_state{conn_state = blocked, rate_limit = Limiter1}
end. end.
run_socket(State = #state{conn_state = blocked}) -> run_socket(State = #client_state{conn_state = blocked}) ->
State; State;
run_socket(State = #state{await_recv = true}) -> run_socket(State = #client_state{await_recv = true}) ->
State; State;
run_socket(State = #state{transport = Transport, socket = Sock}) -> run_socket(State = #client_state{connection = Connection}) ->
Transport:async_recv(Sock, 0, infinity), Connection:async_recv(0, infinity),
State#state{await_recv = true}. State#client_state{await_recv = true}.
noreply(State) ->
{noreply, State, hibernate}.
shutdown(Reason, State) ->
stop({shutdown, Reason}, State).
stop(Reason, State) ->
{stop, Reason, State}.
send_fun(Transport, Sock) ->
fun(Data) ->
try Transport:port_command(Sock, Data) of
true -> ok
catch
error:Error -> exit({socket_error, Error})
end
end.