anlicheng/iot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix emqtt

Browse Source
This commit is contained in:
anlicheng 2023-07-24 14:48:15 +08:00
parent d11c4accca
commit a2ab8f6591
3 changed files with 15 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
apps/iot/include/emqtt.hrl
View File

@ -405,6 +405,12 @@
reason_code = ReasonCode} reason_code = ReasonCode}
}). }).
-define(PUBREC_PACKET(PacketId, ReasonCode),
#mqtt_packet{header = #mqtt_packet_header{type = ?PUBREC},
variable = #mqtt_packet_puback{packet_id = PacketId,
reason_code = ReasonCode}
}).
-define(PUBREC_PACKET(PacketId, ReasonCode, Properties), -define(PUBREC_PACKET(PacketId, ReasonCode, Properties),
#mqtt_packet{header = #mqtt_packet_header{type = ?PUBREC}, #mqtt_packet{header = #mqtt_packet_header{type = ?PUBREC},
variable = #mqtt_packet_puback{packet_id = PacketId, variable = #mqtt_packet_puback{packet_id = PacketId,

4
apps/iot/src/emqtt/emqtt.erl
View File

@ -790,6 +790,10 @@ connected(cast, ?PUBREC_PACKET(PacketId), State = #state{inflight = Inflight}) -
end, end,
send_puback(?PUBREL_PACKET(PacketId), NState); send_puback(?PUBREL_PACKET(PacketId), NState);
connected(cast, ?PUBREC_PACKET(PacketId, ReasonCode), State) ->
lager:notice("[emqtt] Duplicated PUBREC Packet: ~p, reason_code: ~p, client_id: ~p", [PacketId, ReasonCode, State#state.clientid]),
keep_state_and_data;
%%TODO::... if auto_ack is false, should we take PacketId from the map? %%TODO::... if auto_ack is false, should we take PacketId from the map?
connected(cast, ?PUBREL_PACKET(PacketId), connected(cast, ?PUBREL_PACKET(PacketId),
State = #state{awaiting_rel = AwaitingRel, auto_ack = AutoAck}) -> State = #state{awaiting_rel = AwaitingRel, auto_ack = AutoAck}) ->

9
apps/iot/src/emqtt/emqtt_frame.erl
View File

@ -98,8 +98,7 @@ parse_remaining_len(<<>>, Header, Options) ->
parse_remaining_len(Rest, Header, Options) -> parse_remaining_len(Rest, Header, Options) ->
parse_remaining_len(Rest, Header, 1, 0, Options). parse_remaining_len(Rest, Header, 1, 0, Options).
parse_remaining_len(_Bin, _Header, _Multiplier, Length, #{max_size := MaxSize}) parse_remaining_len(_Bin, _Header, _Multiplier, Length, #{max_size := MaxSize}) when Length > MaxSize ->
when Length > MaxSize ->
error(frame_too_large); error(frame_too_large);
parse_remaining_len(<<>>, Header, Multiplier, Length, Options) -> parse_remaining_len(<<>>, Header, Multiplier, Length, Options) ->
{more, fun(Bin) -> parse_remaining_len(Bin, Header, Multiplier, Length, Options) end}; {more, fun(Bin) -> parse_remaining_len(Bin, Header, Multiplier, Length, Options) end};
@ -119,8 +118,10 @@ parse_remaining_len(<<0:1, Len:7, Rest/binary>>, Header, Multiplier, Value,
Options = #{max_size := MaxSize}) -> Options = #{max_size := MaxSize}) ->
FrameLen = Value + Len * Multiplier, FrameLen = Value + Len * Multiplier,
if if
FrameLen > MaxSize -> error(frame_too_large); FrameLen > MaxSize ->
true -> parse_frame(Rest, Header, FrameLen, Options) error(frame_too_large);
true ->
parse_frame(Rest, Header, FrameLen, Options)
end. end.
parse_frame(Bin, Header, 0, Options) -> parse_frame(Bin, Header, 0, Options) ->