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

Merge pull request #14026 from rabbitmq/issue-13785_followup 

MQTT: send acks before disconnecting consumer
This commit is contained in:
Michael Klishin 2025-06-04 15:16:52 +04:00 committed by GitHub
parent 21b6088f00 607b1fda72
commit 74317c76f9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 10 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl vendored
View File

@ -2031,16 +2031,17 @@ handle_queue_event({queue_event, QName, Evt},
State = handle_queue_actions(Actions, State1),
{ok, State};
{eol, Actions} ->
try
State1 = handle_queue_actions(Actions ++ [{queue_down, QName}], State0),
{ConfirmPktIds, U} = rabbit_mqtt_confirms:remove_queue(QName, U0),
QStates = rabbit_queue_type:remove(QName, QStates0),
State = State1#state{queue_states = QStates,
unacked_client_pubs = U},
send_puback(ConfirmPktIds, ?RC_SUCCESS, State),
{ok, State}
State1 = handle_queue_actions(Actions, State0),
{ConfirmPktIds, U} = rabbit_mqtt_confirms:remove_queue(QName, U0),
QStates = rabbit_queue_type:remove(QName, QStates0),
State = State1#state{queue_states = QStates,
unacked_client_pubs = U},
send_puback(ConfirmPktIds, ?RC_SUCCESS, State),
try handle_queue_down(QName, State) of
State2 ->
{ok, State2}
catch throw:consuming_queue_down ->
{error, consuming_queue_down, State0}
{error, consuming_queue_down, State}
end;
{protocol_error, _Type, _Reason, _ReasonArgs} = Error ->
{error, Error, State0}