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

Use the new emqttd_parser API to parse Websocket frame

This commit is contained in:
Feng Lee 2017-02-22 15:43:24 +08:00
parent b4936726fd
commit 88c2b4eaa3
1 changed files with 24 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
src/emqttd_ws.erl
View File

@ -18,13 +18,18 @@
-author("Feng Lee <feng@emqtt.io>"). -author("Feng Lee <feng@emqtt.io>").
-include("emqttd_protocol.hrl").
-import(proplists, [get_value/3]).
-export([handle_request/1, ws_loop/3]). -export([handle_request/1, ws_loop/3]).
%% WebSocket Loop State %% WebSocket Loop State
-record(wsocket_state, {peer, client_pid, packet_opts, parser_fun}). -record(wsocket_state, {peername, client_pid, max_packet_size, parser}).
-define(WSLOG(Level, Peer, Format, Args), -define(WSLOG(Level, Format, Args, State),
lager:Level("WsClient(~s): " ++ Format, [Peer | Args])). lager:Level("WsClient(~s): " ++ Format,
[esockd_net:format(State#wsocket_state.peername) | Args])).
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% Handle WebSocket Request %% Handle WebSocket Request
@ -32,18 +37,14 @@
%% @doc Handle WebSocket Request. %% @doc Handle WebSocket Request.
handle_request(Req) -> handle_request(Req) ->
Peer = Req:get(peer), {ok, Env} = emqttd:env(protocol),
{ok, PktOpts} = emqttd:env(protocol), PacketSize = get_value(max_packet_size, Env, ?MAX_PACKET_SIZE),
ParserFun = emqttd_parser:new(PktOpts), Parser = emqttd_parser:initial_state(PacketSize),
{ReentryWs, ReplyChannel} = upgrade(Req), %% Upgrade WebSocket.
{ReentryWs, ReplyChannel} = mochiweb_websocket:upgrade_connection(Req, fun ?MODULE:ws_loop/3),
{ok, ClientPid} = emqttd_ws_client_sup:start_client(self(), Req, ReplyChannel), {ok, ClientPid} = emqttd_ws_client_sup:start_client(self(), Req, ReplyChannel),
ReentryWs(#wsocket_state{peer = Peer, client_pid = ClientPid, ReentryWs(#wsocket_state{peername = Req:get(peername), parser = Parser,
packet_opts = PktOpts, parser_fun = ParserFun}). max_packet_size = PacketSize, client_pid = ClientPid}).
%% @doc Upgrade WebSocket.
%% @private
upgrade(Req) ->
mochiweb_websocket:upgrade_connection(Req, fun ?MODULE:ws_loop/3).
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% Receive Loop %% Receive Loop
@ -54,25 +55,24 @@ ws_loop(<<>>, State, _ReplyChannel) ->
State; State;
ws_loop([<<>>], State, _ReplyChannel) -> ws_loop([<<>>], State, _ReplyChannel) ->
State; State;
ws_loop(Data, State = #wsocket_state{peer = Peer, client_pid = ClientPid, ws_loop(Data, State = #wsocket_state{client_pid = ClientPid, parser = Parser}, ReplyChannel) ->
parser_fun = ParserFun}, ReplyChannel) -> ?WSLOG(debug, "RECV ~p", [Data], State),
?WSLOG(debug, Peer, "RECV ~p", [Data]),
emqttd_metrics:inc('bytes/received', iolist_size(Data)), emqttd_metrics:inc('bytes/received', iolist_size(Data)),
case catch ParserFun(iolist_to_binary(Data)) of case catch emqttd_parser:parse(iolist_to_binary(Data), Parser) of
{more, NewParser} -> {more, NewParser} ->
State#wsocket_state{parser_fun = NewParser}; State#wsocket_state{parser = NewParser};
{ok, Packet, Rest} -> {ok, Packet, Rest} ->
gen_server:cast(ClientPid, {received, Packet}), gen_server:cast(ClientPid, {received, Packet}),
ws_loop(Rest, reset_parser(State), ReplyChannel); ws_loop(Rest, reset_parser(State), ReplyChannel);
{error, Error} -> {error, Error} ->
?WSLOG(error, Peer, "Frame error: ~p", [Error]), ?WSLOG(error, "Frame error: ~p", [Error], State),
exit({shutdown, Error}); exit({shutdown, Error});
{'EXIT', Reason} -> {'EXIT', Reason} ->
?WSLOG(error, Peer, "Frame error: ~p", [Reason]), ?WSLOG(error, "Frame error: ~p", [Reason], State),
?WSLOG(error, Peer, "Error data: ~p", [Data]), ?WSLOG(error, "Error data: ~p", [Data], State),
exit({shutdown, parser_error}) exit({shutdown, parser_error})
end. end.
reset_parser(State = #wsocket_state{packet_opts = PktOpts}) -> reset_parser(State = #wsocket_state{max_packet_size = PacketSize}) ->
State#wsocket_state{parser_fun = emqttd_parser:new(PktOpts)}. State#wsocket_state{parser = emqttd_parser:initial_state(PacketSize)}.