fix(emqx_rule_engine): set inc_action_metrics as async_reply_fun #11126
Conversation
sstrigler
force-pushed
the
EMQX-8842-fix-rule-metrics
branch
2 times, most recently
from
896890a
to
778576b
Compare
There was a problem hiding this comment.
LGTM. The only comments that I have are the things we discussed before. Maybe async_reply_fun is not the optimal name for the field as we use it even when the "query" is not async. Maybe on_complete_reply_fun would be better? Is async_reply_fun used by something else? Could it be "overwritten" somewhere in the call chain? Also as we discussed before, it would be good with a test case.
apps/emqx_bridge/src/emqx_bridge.erl
Outdated
| @@ -220,14 +220,15 @@ send_to_matched_egress_bridges(Topic, Msg) -> | |||
| send_message(BridgeId, Message) -> | |||
| {BridgeType, BridgeName} = emqx_bridge_resource:parse_bridge_id(BridgeId), | |||
| ResId = emqx_bridge_resource:resource_id(BridgeType, BridgeName), | |||
| send_message(BridgeType, BridgeName, ResId, Message). | |||
| send_message(BridgeType, BridgeName, ResId, Message, undefined). | |||
There was a problem hiding this comment.
Nit:
| send_message(BridgeType, BridgeName, ResId, Message, undefined). | |
| Opts = #{reply_to => undefined}, | |
| send_message(BridgeType, BridgeName, ResId, Message, Opts). |
| @@ -158,7 +158,9 @@ simple_async_query(Id, Request, QueryOpts0) -> | |||
| Ref = make_request_ref(), | |||
| Result = call_query(async_if_possible, Id, Index, Ref, ?SIMPLE_QUERY(Request), QueryOpts), | |||
| _ = handle_query_result(Id, Result, _HasBeenSent = false), | |||
| Result. | |||
| maybe_apply_async_reply_fun( | |||
There was a problem hiding this comment.
I think we should leave the calling of the reply function to do_reply_caller, and just inject the callback as the From here instead of undefined.
i.e., something like:
ReplyTo = maps:get(reply_to, QueryOpts, undefined),
Query = ?QUERY(ReplyTo, Request, _HasBeenSeen = false, _ExpireAt = infinity),
...and then I think do_reply_caller will call {Fun, Args} from From without further changes.
sstrigler
force-pushed
the
EMQX-8842-fix-rule-metrics
branch
from
adb352a
to
ce09a9a
Compare
sstrigler
force-pushed
the
EMQX-8842-fix-rule-metrics
branch
from
ce09a9a
to
3d91098
Compare
| actions => [BridgeId] | ||
| } | ||
| ), | ||
| timer:sleep(100), |
There was a problem hiding this comment.
minor comment: to avoid the risk of creating flake test and not making the tests slow to run one could loop until the the expected condition is met and fail if it is not met after a certain amount of time (see https://github.com/klarna/snabbkaffe/blob/master/src/snabbkaffe.erl#L313)
| emqx_resource_metrics:matched_inc(Id), | ||
| Ref = make_request_ref(), | ||
| Result = call_query(force_sync, Id, Index, Ref, ?SIMPLE_QUERY(Request), QueryOpts), | ||
| ReplyTo = maps:get(reply_to, QueryOpts0, undefined), |
There was a problem hiding this comment.
Nit: now we have 2 different query option keys to specify async reply functions. Maybe we could choose only one of async_reply_fun and reply_to for this query option to avoid confusion?
|
I鈥檒l create a follow-up ticket |
Fixes EMQX-8842
Next try. This time using a callback.
Summary
馃 Generated by Copilot at 5c23ed7
Added support for asynchronous replies from bridge resources in rule engine and resource buffer. This allows bridge actions to update metrics and tracing data after sending or receiving messages. Modified
emqx_bridge,emqx_resource_buffer_worker, andemqx_rule_runtimemodules to implement this feature.PR Checklist
Please convert it to a draft if any of the following conditions are not met. Reviewers may skip over until all the items are checked:
changes/{ce,ee}/(feat|perf|fix)-<PR-id>.en.mdfilesIf there should be document changes, a PR to emqx-docs.git is sent, or a jira ticket is created to follow upSchema changes are backward compatible