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

fix(stomp): Fix dialyzer warnings

This commit is contained in:
Zaiming Shi 2020-11-04 13:07:42 +01:00
parent bcae92ae6d
commit 760ef9210a
3 changed files with 15 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
apps/emqx_stomp/include/emqx_stomp.hrl
View File

@ -24,7 +24,7 @@
%% STOMP Frame
%%--------------------------------------------------------------------
-record(stomp_frame, {command, headers = [], body = <<>> :: iolist()}).
-record(stomp_frame, {command, headers = [], body = <<>> :: iodata()}).
-type(stomp_frame() :: #stomp_frame{}).

6
apps/emqx_stomp/src/emqx_stomp_frame.erl
View File

@ -87,7 +87,11 @@
-define(IS_ESC(Ch), Ch == ?CR; Ch == ?LF; Ch == ?BSL; Ch == ?COLON).
-record(parser_state, {cmd, headers = [], hdname, acc = <<>>, limit}).
-record(parser_state, {cmd,
headers = [],
hdname,
acc = <<>> :: binary(),
limit}).
-record(frame_limit, {max_header_num, max_header_length, max_body_length}).

17
apps/emqx_stomp/src/emqx_stomp_protocol.erl
View File

@ -99,11 +99,11 @@ received(#stomp_frame{command = <<"CONNECT">>, headers = Headers},
send(connected_frame([{<<"version">>, Version},
{<<"heart-beat">>, reverse_heartbeats(Heartbeats)}]), NewState);
false ->
send(error_frame(undefined, <<"Login or passcode error!">>), State),
_ = send(error_frame(undefined, <<"Login or passcode error!">>), State),
{error, login_or_passcode_error, State}
end;
{error, Msg} ->
send(error_frame([{<<"version">>, <<"1.0,1.1,1.2">>},
_ = send(error_frame([{<<"version">>, <<"1.0,1.1,1.2">>},
{<<"content-type">>, <<"text/plain">>}], undefined, Msg), State),
{error, unsupported_version, State}
end;
@ -114,10 +114,9 @@ received(#stomp_frame{command = <<"CONNECT">>}, State = #stomp_proto{connected =
received(#stomp_frame{command = <<"SEND">>, headers = Headers, body = Body}, State) ->
Topic = header(<<"destination">>, Headers),
Action = fun(State0) ->
maybe_send_receipt(receipt_id(Headers), State0),
emqx_broker:publish(
make_mqtt_message(Topic, Headers, iolist_to_binary(Body))
),
_ = maybe_send_receipt(receipt_id(Headers), State0),
_ = emqx_broker:publish(
make_mqtt_message(Topic, Headers, iolist_to_binary(Body))),
State0
end,
case header(<<"transaction">>, Headers) of
@ -160,7 +159,7 @@ received(#stomp_frame{command = <<"UNSUBSCRIBE">>, headers = Headers},
received(#stomp_frame{command = <<"ACK">>, headers = Headers}, State) ->
Id = header(<<"id">>, Headers),
Action = fun(State0) ->
maybe_send_receipt(receipt_id(Headers), State0),
_ = maybe_send_receipt(receipt_id(Headers), State0),
ack(Id, State0)
end,
case header(<<"transaction">>, Headers) of
@ -176,7 +175,7 @@ received(#stomp_frame{command = <<"ACK">>, headers = Headers}, State) ->
received(#stomp_frame{command = <<"NACK">>, headers = Headers}, State) ->
Id = header(<<"id">>, Headers),
Action = fun(State0) ->
maybe_send_receipt(receipt_id(Headers), State0),
_ = maybe_send_receipt(receipt_id(Headers), State0),
nack(Id, State0)
end,
case header(<<"transaction">>, Headers) of
@ -226,7 +225,7 @@ received(#stomp_frame{command = <<"ABORT">>, headers = Headers}, State) ->
end;
received(#stomp_frame{command = <<"DISCONNECT">>, headers = Headers}, State) ->
maybe_send_receipt(receipt_id(Headers), State),
_ = maybe_send_receipt(receipt_id(Headers), State),
{stop, normal, State}.
send(Msg = #message{topic = Topic, headers = Headers, payload = Payload},