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

Ensure the 'CONNACK' and 'DISCONNECT' packets delivered.

This commit is contained in:
Feng Lee 2019-08-09 15:01:39 +08:00
parent 52a85d9bb9
commit 03744ead30
3 changed files with 12 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/emqx_channel.erl
View File

@ -516,6 +516,9 @@ handle_incoming(Packet = ?PACKET(Type), SuccFun,
State#state{proto_state = NProtoState});
{error, Reason, NProtoState} ->
shutdown(Reason, State#state{proto_state = NProtoState});
{error, Reason, OutPacket, NProtoState} ->
Shutdown = fun(NewSt) -> shutdown(Reason, NewSt) end,
handle_outgoing(OutPacket, Shutdown, State#state{proto_state = NProtoState});
{stop, Error, NProtoState} ->
stop(Error, State#state{proto_state = NProtoState})
end.

10
src/emqx_protocol.erl
View File

@ -416,11 +416,15 @@ handle_out({unsuback, PacketId, ReasonCodes}, PState = #protocol{proto_ver = ?MQ
handle_out({unsuback, PacketId, _ReasonCodes}, PState) ->
{ok, ?UNSUBACK_PACKET(PacketId), PState};
handle_out({disconnect, ReasonCode}, PState) ->
{ok, PState};
handle_out({disconnect, ReasonCode}, PState = #protocol{proto_ver = ?MQTT_PROTO_V5}) ->
Reason = emqx_reason_codes:name(ReasonCode),
{error, Reason, ?DISCONNECT_PACKET(ReasonCode), PState};
handle_out({disconnect, ReasonCode}, PState = #protocol{proto_ver = ProtoVer}) ->
{error, emqx_reason_codes:name(ReasonCode, ProtoVer), PState};
handle_out(Packet, PState) ->
io:format("Out: ~p~n", [Packet]),
?LOG(error, "Unexpected out:~p", [Packet]),
{ok, PState}.
%%--------------------------------------------------------------------

2
src/emqx_ws_channel.erl
View File

@ -436,6 +436,8 @@ handle_incoming(Packet = ?PACKET(Type), SuccFun,
SuccFun(enqueue(OutPackets, State#state{proto_state = NProtoState}));
{error, Reason, NProtoState} ->
stop(Reason, State#state{proto_state = NProtoState});
{error, Reason, OutPacket, NProtoState} ->
stop(Reason, enqueue(OutPacket, State#state{proto_state = NProtoState}));
{stop, Error, NProtoState} ->
stop(Error, State#state{proto_state = NProtoState})
end.