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

perf: no need to format event message if no bridge matched

This commit is contained in:
Zaiming (Stone) Shi 2023-10-31 16:46:36 +01:00
parent 848f9c8dd6
commit db5cdb6742
1 changed files with 40 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

74
apps/emqx_bridge/src/emqx_bridge.erl
View File

@ -191,45 +191,50 @@ unload_hook() ->
on_message_publish(Message = #message{topic = Topic, flags = Flags}) -> on_message_publish(Message = #message{topic = Topic, flags = Flags}) ->
case maps:get(sys, Flags, false) of case maps:get(sys, Flags, false) of
false -> false ->
{Msg, _} = emqx_rule_events:eventmsg_publish(Message), send_to_matched_egress_bridges(Topic, Message);
send_to_matched_egress_bridges(Topic, Msg);
true -> true ->
ok ok
end, end,
{ok, Message}. {ok, Message}.
send_to_matched_egress_bridges(Topic, Msg) -> send_to_matched_egress_bridges(Topic, Message) ->
MatchedBridgeIds = get_matched_egress_bridges(Topic), case get_matched_egress_bridges(Topic) of
lists:foreach( [] ->
fun(Id) -> ok;
try send_message(Id, Msg) of Ids ->
{error, Reason} -> {Msg, _} = emqx_rule_events:eventmsg_publish(Message),
?SLOG(error, #{ send_to_matched_egress_bridges_loop(Topic, Msg, Ids)
msg => "send_message_to_bridge_failed", end.
bridge => Id,
error => Reason send_to_matched_egress_bridges_loop(_Topic, _Msg, []) ->
}); ok;
_ -> send_to_matched_egress_bridges_loop(Topic, Msg, [Id | Ids]) ->
ok try send_message(Id, Msg) of
catch {error, Reason} ->
throw:Reason -> ?SLOG(error, #{
?SLOG(error, #{ msg => "send_message_to_bridge_failed",
msg => "send_message_to_bridge_exception", bridge => Id,
bridge => Id, error => Reason
reason => emqx_utils:redact(Reason) });
}); _ ->
Err:Reason:ST -> ok
?SLOG(error, #{ catch
msg => "send_message_to_bridge_exception", throw:Reason ->
bridge => Id, ?SLOG(error, #{
error => Err, msg => "send_message_to_bridge_exception",
reason => emqx_utils:redact(Reason), bridge => Id,
stacktrace => emqx_utils:redact(ST) reason => emqx_utils:redact(Reason)
}) });
end Err:Reason:ST ->
end, ?SLOG(error, #{
MatchedBridgeIds msg => "send_message_to_bridge_exception",
). bridge => Id,
error => Err,
reason => emqx_utils:redact(Reason),
stacktrace => emqx_utils:redact(ST)
})
end,
send_to_matched_egress_bridges_loop(Topic, Msg, Ids).
send_message(BridgeId, Message) -> send_message(BridgeId, Message) ->
{BridgeType, BridgeName} = emqx_bridge_resource:parse_bridge_id(BridgeId), {BridgeType, BridgeName} = emqx_bridge_resource:parse_bridge_id(BridgeId),
@ -571,6 +576,7 @@ flatten_confs(Conf0) ->
do_flatten_confs(Type, Conf0) -> do_flatten_confs(Type, Conf0) ->
[{{Type, Name}, Conf} || {Name, Conf} <- maps:to_list(Conf0)]. [{{Type, Name}, Conf} || {Name, Conf} <- maps:to_list(Conf0)].
%% TODO: create a topic index for this
get_matched_egress_bridges(Topic) -> get_matched_egress_bridges(Topic) ->
Bridges = emqx:get_config([bridges], #{}), Bridges = emqx:get_config([bridges], #{}),
maps:fold( maps:fold(