fix: run `message.dropped` hook, inc `messages.dropped` metrics
- when awaiting_rel full - packet identifier in use (QoS2 packet resend)
This commit is contained in:
parent
c93568063d
commit
b88398c3c6
|
@ -752,15 +752,6 @@ do_publish(
|
||||||
ok = emqx_metrics:inc('packets.publish.inuse'),
|
ok = emqx_metrics:inc('packets.publish.inuse'),
|
||||||
handle_out(pubrec, {PacketId, RC}, Channel);
|
handle_out(pubrec, {PacketId, RC}, Channel);
|
||||||
{error, RC = ?RC_RECEIVE_MAXIMUM_EXCEEDED} ->
|
{error, RC = ?RC_RECEIVE_MAXIMUM_EXCEEDED} ->
|
||||||
?SLOG(
|
|
||||||
warning,
|
|
||||||
#{
|
|
||||||
msg => "dropped_qos2_packet",
|
|
||||||
reason => emqx_reason_codes:name(RC),
|
|
||||||
packet_id => PacketId
|
|
||||||
},
|
|
||||||
#{topic => Msg#message.topic}
|
|
||||||
),
|
|
||||||
ok = emqx_metrics:inc('packets.publish.dropped'),
|
ok = emqx_metrics:inc('packets.publish.dropped'),
|
||||||
handle_out(disconnect, RC, Channel)
|
handle_out(disconnect, RC, Channel)
|
||||||
end.
|
end.
|
||||||
|
|
|
@ -390,15 +390,29 @@ publish(
|
||||||
AwaitingRel1 = maps:put(PacketId, Ts, AwaitingRel),
|
AwaitingRel1 = maps:put(PacketId, Ts, AwaitingRel),
|
||||||
{ok, Results, Session#session{awaiting_rel = AwaitingRel1}};
|
{ok, Results, Session#session{awaiting_rel = AwaitingRel1}};
|
||||||
true ->
|
true ->
|
||||||
{error, ?RC_PACKET_IDENTIFIER_IN_USE}
|
drop_qos2_msg(PacketId, Msg, ?RC_PACKET_IDENTIFIER_IN_USE)
|
||||||
end;
|
end;
|
||||||
true ->
|
true ->
|
||||||
{error, ?RC_RECEIVE_MAXIMUM_EXCEEDED}
|
drop_qos2_msg(PacketId, Msg, ?RC_RECEIVE_MAXIMUM_EXCEEDED)
|
||||||
end;
|
end;
|
||||||
%% Publish QoS0/1 directly
|
%% Publish QoS0/1 directly
|
||||||
publish(_ClientInfo, _PacketId, Msg, Session) ->
|
publish(_ClientInfo, _PacketId, Msg, Session) ->
|
||||||
{ok, emqx_broker:publish(Msg), Session}.
|
{ok, emqx_broker:publish(Msg), Session}.
|
||||||
|
|
||||||
|
drop_qos2_msg(PacketId, Msg, RC) ->
|
||||||
|
?SLOG(
|
||||||
|
warning,
|
||||||
|
#{
|
||||||
|
msg => "dropped_qos2_packet",
|
||||||
|
reason => emqx_reason_codes:name(RC),
|
||||||
|
packet_id => PacketId
|
||||||
|
},
|
||||||
|
#{topic => Msg#message.topic}
|
||||||
|
),
|
||||||
|
ok = emqx_metrics:inc('messages.dropped'),
|
||||||
|
ok = emqx_hooks:run('message.dropped', [Msg, #{node => node()}, emqx_reason_codes:name(RC)]),
|
||||||
|
{error, RC}.
|
||||||
|
|
||||||
is_awaiting_full(#session{max_awaiting_rel = infinity}) ->
|
is_awaiting_full(#session{max_awaiting_rel = infinity}) ->
|
||||||
false;
|
false;
|
||||||
is_awaiting_full(#session{
|
is_awaiting_full(#session{
|
||||||
|
|
|
@ -11,3 +11,5 @@
|
||||||
- Refactor: use `POST` not `PUT` for `/users/{name}/change_pwd` [#9533](https://github.com/emqx/emqx/pull/9533).
|
- Refactor: use `POST` not `PUT` for `/users/{name}/change_pwd` [#9533](https://github.com/emqx/emqx/pull/9533).
|
||||||
|
|
||||||
## Bug fixes
|
## Bug fixes
|
||||||
|
|
||||||
|
- Trigger `message.dropped` hook when QoS2 message is resend by client with a same packet id, or 'awaiting_rel' queue is full [#9487](https://github.com/emqx/emqx/pull/9487).
|
||||||
|
|
|
@ -11,3 +11,5 @@
|
||||||
- 重构: `/users/{name}/change_pwd` 的请求方式从 `PUT` 改为了 `POST` [#9533](https://github.com/emqx/emqx/pull/9533)。
|
- 重构: `/users/{name}/change_pwd` 的请求方式从 `PUT` 改为了 `POST` [#9533](https://github.com/emqx/emqx/pull/9533)。
|
||||||
|
|
||||||
## 修复
|
## 修复
|
||||||
|
|
||||||
|
- 当 QoS2 消息被重发(使用相同 Packet ID),或当 'awaiting_rel' 队列已满时,触发消息丢弃钩子(`message.dropped`)及计数器 [#9487](https://github.com/emqx/emqx/pull/9487)。
|
||||||
|
|
Loading…
Reference in New Issue