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

fix: STOMP heartbeat

This commit is contained in:
chengshq 2024-04-24 01:42:26 +08:00
parent 5804b70c1f
commit 75dab4dff0
4 changed files with 17 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
apps/emqx_gateway/src/bhvrs/emqx_gateway_conn.erl
View File

@ -710,7 +710,10 @@ parse_incoming(
) -> ) ->
try FrameMod:parse(Data, ParseState) of try FrameMod:parse(Data, ParseState) of
{more, NParseState} -> {more, NParseState} ->
{Packets, State#state{parse_state = NParseState}}; if
Data == <<$\n>> -> {[Data], State#state{parse_state = NParseState}};
true -> {Packets, State#state{parse_state = NParseState}}
end;
{ok, Packet, Rest, NParseState} -> {ok, Packet, Rest, NParseState} ->
NState = State#state{parse_state = NParseState}, NState = State#state{parse_state = NParseState},
parse_incoming(Rest, [Packet | Packets], NState) parse_incoming(Rest, [Packet | Packets], NState)

22
apps/emqx_gateway_stomp/src/emqx_stomp_channel.erl
View File

@ -653,6 +653,12 @@ handle_in(
] ]
end, end,
{ok, Outgoings, Channel}; {ok, Outgoings, Channel};
handle_in(
<<$\n>>,
Channel = #channel{heartbeat = Heartbeat}
) ->
NewHeartbeat = emqx_stomp_heartbeat:reset(incoming, 0, Heartbeat),
{ok, Channel#channel{heartbeat = NewHeartbeat}};
handle_in({frame_error, Reason}, Channel = #channel{conn_state = idle}) -> handle_in({frame_error, Reason}, Channel = #channel{conn_state = idle}) ->
shutdown(Reason, Channel); shutdown(Reason, Channel);
handle_in({frame_error, Reason}, Channel = #channel{conn_state = _ConnState}) -> handle_in({frame_error, Reason}, Channel = #channel{conn_state = _ConnState}) ->
@ -1116,19 +1122,9 @@ handle_timeout(
{keepalive_send, NewVal}, {keepalive_send, NewVal},
Channel = #channel{heartbeat = HrtBt} Channel = #channel{heartbeat = HrtBt}
) -> ) ->
case emqx_stomp_heartbeat:check(outgoing, NewVal, HrtBt) of NHrtBt = emqx_stomp_heartbeat:reset(outgoing, NewVal, HrtBt),
{error, timeout} -> NChannel = Channel#channel{heartbeat = NHrtBt},
NHrtBt = emqx_stomp_heartbeat:reset(outgoing, NewVal, HrtBt), {ok, {outgoing, emqx_stomp_frame:make(?CMD_HEARTBEAT)}, reset_timer(outgoing_timer, NChannel)};
NChannel = Channel#channel{heartbeat = NHrtBt},
{ok, {outgoing, emqx_stomp_frame:make(?CMD_HEARTBEAT)},
reset_timer(outgoing_timer, NChannel)};
{ok, NHrtBt} ->
{ok,
reset_timer(
outgoing_timer,
Channel#channel{heartbeat = NHrtBt}
)}
end;
handle_timeout(_TRef, clean_trans, Channel = #channel{transaction = Trans}) -> handle_timeout(_TRef, clean_trans, Channel = #channel{transaction = Trans}) ->
Now = erlang:system_time(millisecond), Now = erlang:system_time(millisecond),
NTrans = maps:filter( NTrans = maps:filter(

4
apps/emqx_gateway_stomp/src/emqx_stomp_frame.erl
View File

@ -346,7 +346,9 @@ serialize_pkt(
serialize_pkt(header, {Name, Val}) when is_integer(Val) -> serialize_pkt(header, {Name, Val}) when is_integer(Val) ->
[escape(Name), ?COLON, integer_to_list(Val), ?LF]; [escape(Name), ?COLON, integer_to_list(Val), ?LF];
serialize_pkt(header, {Name, Val}) -> serialize_pkt(header, {Name, Val}) ->
[escape(Name), ?COLON, escape(Val), ?LF]. [escape(Name), ?COLON, escape(Val), ?LF];
serialize_pkt(<<$\n>>, _SerializeOpts) ->
<<$\n>>.
escape(Bin) when is_binary(Bin) -> escape(Bin) when is_binary(Bin) ->
<<<<(escape(Ch))/binary>> || <<Ch>> <= Bin>>; <<<<(escape(Ch))/binary>> || <<Ch>> <= Bin>>;

2
apps/emqx_gateway_stomp/test/emqx_stomp_SUITE.erl
View File

@ -153,7 +153,7 @@ t_heartbeat(_) ->
{<<"host">>, <<"127.0.0.1:61613">>}, {<<"host">>, <<"127.0.0.1:61613">>},
{<<"login">>, <<"guest">>}, {<<"login">>, <<"guest">>},
{<<"passcode">>, <<"guest">>}, {<<"passcode">>, <<"guest">>},
{<<"heart-beat">>, <<"1000,800">>} {<<"heart-beat">>, <<"500,800">>}
] ]
) )
), ),