root
/
rabbitmq-server
mirror of https://github.com/rabbitmq/rabbitmq-server.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Merge pull request #11683 from rabbitmq/mergify/bp/v3.13.x/pr-11682 

Catch abrupt TCP closure when processing `queue_event` (backport #11676) (backport #11682)
This commit is contained in:
Michael Klishin 2024-07-11 17:30:23 -04:00 committed by GitHub
parent 0e8ea17d20 6fb22af96d
commit c10364af6e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 22 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
deps/rabbitmq_mqtt/src/rabbit_mqtt_reader.erl vendored
View File

@ -140,11 +140,15 @@ handle_cast({close_connection, Reason},
handle_cast(QueueEvent = {queue_event, _, _},
State = #state{proc_state = PState0}) ->
case rabbit_mqtt_processor:handle_queue_event(QueueEvent, PState0) of
{ok, PState} ->
maybe_process_deferred_recv(control_throttle(pstate(State, PState)));
{error, Reason, PState} ->
{stop, Reason, pstate(State, PState)}
try
case rabbit_mqtt_processor:handle_queue_event(QueueEvent, PState0) of
{ok, PState} ->
maybe_process_deferred_recv(control_throttle(pstate(State, PState)));
{error, Reason0, PState} ->
{stop, Reason0, pstate(State, PState)}
end
catch throw:{send_failed, Reason1} ->
network_error(Reason1, State)
end;
handle_cast({force_event_refresh, Ref}, State0) ->
@ -328,17 +332,17 @@ process_received_bytes(Bytes, State = #state{socket = Socket,
{ok, Packet, Rest, ParseState1} ->
case ProcState of
connect_packet_unprocessed ->
Send = fun(Data) ->
case rabbit_net:send(Socket, Data) of
ok ->
ok;
{error, Reason} ->
?LOG_ERROR("writing to MQTT socket ~p failed: ~p",
[Socket, Reason]),
exit({send_failed, Reason})
end
end,
try rabbit_mqtt_processor:init(Packet, Socket, ConnName, Send) of
SendFun = fun(Data) ->
case rabbit_net:send(Socket, Data) of
ok ->
ok;
{error, Reason} ->
?LOG_ERROR("writing to MQTT socket ~p failed: ~p",
[Socket, Reason]),
throw({send_failed, Reason})
end
end,
try rabbit_mqtt_processor:init(Packet, Socket, ConnName, SendFun) of
{ok, ProcState1} ->
?LOG_INFO("Accepted MQTT connection ~ts for client ID ~ts",
[ConnName, rabbit_mqtt_processor:info(client_id, ProcState1)]),
@ -354,7 +358,7 @@ process_received_bytes(Bytes, State = #state{socket = Socket,
?LOG_ERROR("Rejected MQTT connection ~ts with Connect Reason Code ~p",
[ConnName, ConnectReasonCode]),
{stop, shutdown, {_SendWill = false, State}}
catch exit:{send_failed, Reason} ->
catch throw:{send_failed, Reason} ->
network_error(Reason, State)
end;
_ ->
@ -375,7 +379,7 @@ process_received_bytes(Bytes, State = #state{socket = Socket,
{stop, {shutdown, Reason}, pstate(State, ProcState1)};
{stop, {disconnect, {client_initiated, SendWill}}, ProcState1} ->
{stop, normal, {SendWill, pstate(State, ProcState1)}}
catch exit:{send_failed, Reason} ->
catch throw:{send_failed, Reason} ->
network_error(Reason, State)
end
end;