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

Merge pull request #9422 from emqx/1124-refactor-avoid-creating-atom-at-run 

refactor: use static function references
This commit is contained in:
Zaiming (Stone) Shi 2022-12-06 14:24:05 +01:00 committed by GitHub
parent 3e29e37691 cca3421308
commit 380b75d4a9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 136 additions and 112 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

122
apps/emqx_rule_engine/src/emqx_rule_api_schema.erl
View File

@ -65,10 +65,7 @@ fields("rule_info") ->
] ++ fields("rule_creation"); ] ++ fields("rule_creation");
%% TODO: we can delete this API if the Dashboard not depends on it %% TODO: we can delete this API if the Dashboard not depends on it
fields("rule_events") -> fields("rule_events") ->
ETopics = [ ETopics = emqx_rule_events:event_topics_enum(),
binary_to_atom(emqx_rule_events:event_topic(E))
|| E <- emqx_rule_events:event_names()
],
[ [
{"event", sc(hoconsc:enum(ETopics), #{desc => ?DESC("rs_event"), required => true})}, {"event", sc(hoconsc:enum(ETopics), #{desc => ?DESC("rs_event"), required => true})},
{"title", sc(binary(), #{desc => ?DESC("rs_title"), example => "some title"})}, {"title", sc(binary(), #{desc => ?DESC("rs_title"), example => "some title"})},
@ -150,77 +147,43 @@ fields("node_metrics") ->
fields("metrics"); fields("metrics");
fields("ctx_pub") -> fields("ctx_pub") ->
[ [
{"event_type", sc(message_publish, #{desc => ?DESC("event_event_type"), required => true})}, {"event_type", event_type_sc(message_publish)},
{"id", sc(binary(), #{desc => ?DESC("event_id")})}, {"id", sc(binary(), #{desc => ?DESC("event_id")})}
{"clientid", sc(binary(), #{desc => ?DESC("event_clientid")})}, | msg_event_common_fields()
{"username", sc(binary(), #{desc => ?DESC("event_username")})}, ];
{"payload", sc(binary(), #{desc => ?DESC("event_payload")})},
{"peerhost", sc(binary(), #{desc => ?DESC("event_peerhost")})},
{"topic", sc(binary(), #{desc => ?DESC("event_topic")})},
{"publish_received_at",
sc(integer(), #{
desc => ?DESC("event_publish_received_at")
})}
] ++ [qos()];
fields("ctx_sub") -> fields("ctx_sub") ->
[ [
{"event_type", {"event_type", event_type_sc(session_subscribed)}
sc(session_subscribed, #{desc => ?DESC("event_event_type"), required => true})}, | msg_event_common_fields()
{"clientid", sc(binary(), #{desc => ?DESC("event_clientid")})}, ];
{"username", sc(binary(), #{desc => ?DESC("event_username")})},
{"payload", sc(binary(), #{desc => ?DESC("event_payload")})},
{"peerhost", sc(binary(), #{desc => ?DESC("event_peerhost")})},
{"topic", sc(binary(), #{desc => ?DESC("event_topic")})},
{"publish_received_at",
sc(integer(), #{
desc => ?DESC("event_publish_received_at")
})}
] ++ [qos()];
fields("ctx_unsub") -> fields("ctx_unsub") ->
[ [
{"event_type", {"event_type", event_type_sc(session_unsubscribed)}
sc(session_unsubscribed, #{desc => ?DESC("event_event_type"), required => true})} | proplists:delete("event_type", fields("ctx_sub"))
] ++ ];
proplists:delete("event_type", fields("ctx_sub"));
fields("ctx_delivered") -> fields("ctx_delivered") ->
[ [
{"event_type", {"event_type", event_type_sc(message_delivered)},
sc(message_delivered, #{desc => ?DESC("event_event_type"), required => true})},
{"id", sc(binary(), #{desc => ?DESC("event_id")})}, {"id", sc(binary(), #{desc => ?DESC("event_id")})},
{"from_clientid", sc(binary(), #{desc => ?DESC("event_from_clientid")})}, {"from_clientid", sc(binary(), #{desc => ?DESC("event_from_clientid")})},
{"from_username", sc(binary(), #{desc => ?DESC("event_from_username")})}, {"from_username", sc(binary(), #{desc => ?DESC("event_from_username")})}
{"clientid", sc(binary(), #{desc => ?DESC("event_clientid")})}, | msg_event_common_fields()
{"username", sc(binary(), #{desc => ?DESC("event_username")})}, ];
{"payload", sc(binary(), #{desc => ?DESC("event_payload")})},
{"peerhost", sc(binary(), #{desc => ?DESC("event_peerhost")})},
{"topic", sc(binary(), #{desc => ?DESC("event_topic")})},
{"publish_received_at",
sc(integer(), #{
desc => ?DESC("event_publish_received_at")
})}
] ++ [qos()];
fields("ctx_acked") -> fields("ctx_acked") ->
[{"event_type", sc(message_acked, #{desc => ?DESC("event_event_type"), required => true})}] ++ [
proplists:delete("event_type", fields("ctx_delivered")); {"event_type", event_type_sc(message_acked)}
| proplists:delete("event_type", fields("ctx_delivered"))
];
fields("ctx_dropped") -> fields("ctx_dropped") ->
[ [
{"event_type", sc(message_dropped, #{desc => ?DESC("event_event_type"), required => true})}, {"event_type", event_type_sc(message_dropped)},
{"id", sc(binary(), #{desc => ?DESC("event_id")})}, {"id", sc(binary(), #{desc => ?DESC("event_id")})},
{"reason", sc(binary(), #{desc => ?DESC("event_ctx_dropped")})}, {"reason", sc(binary(), #{desc => ?DESC("event_ctx_dropped")})}
{"clientid", sc(binary(), #{desc => ?DESC("event_clientid")})}, | msg_event_common_fields()
{"username", sc(binary(), #{desc => ?DESC("event_username")})}, ];
{"payload", sc(binary(), #{desc => ?DESC("event_payload")})},
{"peerhost", sc(binary(), #{desc => ?DESC("event_peerhost")})},
{"topic", sc(binary(), #{desc => ?DESC("event_topic")})},
{"publish_received_at",
sc(integer(), #{
desc => ?DESC("event_publish_received_at")
})}
] ++ [qos()];
fields("ctx_connected") -> fields("ctx_connected") ->
[ [
{"event_type", {"event_type", event_type_sc(client_connected)},
sc(client_connected, #{desc => ?DESC("event_event_type"), required => true})},
{"clientid", sc(binary(), #{desc => ?DESC("event_clientid")})}, {"clientid", sc(binary(), #{desc => ?DESC("event_clientid")})},
{"username", sc(binary(), #{desc => ?DESC("event_username")})}, {"username", sc(binary(), #{desc => ?DESC("event_username")})},
{"mountpoint", sc(binary(), #{desc => ?DESC("event_mountpoint")})}, {"mountpoint", sc(binary(), #{desc => ?DESC("event_mountpoint")})},
@ -239,8 +202,7 @@ fields("ctx_connected") ->
]; ];
fields("ctx_disconnected") -> fields("ctx_disconnected") ->
[ [
{"event_type", {"event_type", event_type_sc(client_disconnected)},
sc(client_disconnected, #{desc => ?DESC("event_event_type"), required => true})},
{"clientid", sc(binary(), #{desc => ?DESC("event_clientid")})}, {"clientid", sc(binary(), #{desc => ?DESC("event_clientid")})},
{"username", sc(binary(), #{desc => ?DESC("event_username")})}, {"username", sc(binary(), #{desc => ?DESC("event_username")})},
{"reason", sc(binary(), #{desc => ?DESC("event_ctx_disconnected_reason")})}, {"reason", sc(binary(), #{desc => ?DESC("event_ctx_disconnected_reason")})},
@ -253,7 +215,7 @@ fields("ctx_disconnected") ->
]; ];
fields("ctx_connack") -> fields("ctx_connack") ->
[ [
{"event_type", sc(client_connack, #{desc => ?DESC("event_event_type"), required => true})}, {"event_type", event_type_sc(client_connack)},
{"reason_code", sc(binary(), #{desc => ?DESC("event_ctx_connack_reason_code")})}, {"reason_code", sc(binary(), #{desc => ?DESC("event_ctx_connack_reason_code")})},
{"clientid", sc(binary(), #{desc => ?DESC("event_clientid")})}, {"clientid", sc(binary(), #{desc => ?DESC("event_clientid")})},
{"clean_start", sc(boolean(), #{desc => ?DESC("event_clean_start"), default => true})}, {"clean_start", sc(boolean(), #{desc => ?DESC("event_clean_start"), default => true})},
@ -271,8 +233,7 @@ fields("ctx_connack") ->
]; ];
fields("ctx_check_authz_complete") -> fields("ctx_check_authz_complete") ->
[ [
{"event_type", {"event_type", event_type_sc(client_check_authz_complete)},
sc(client_check_authz_complete, #{desc => ?DESC("event_event_type"), required => true})},
{"clientid", sc(binary(), #{desc => ?DESC("event_clientid")})}, {"clientid", sc(binary(), #{desc => ?DESC("event_clientid")})},
{"username", sc(binary(), #{desc => ?DESC("event_username")})}, {"username", sc(binary(), #{desc => ?DESC("event_username")})},
{"peerhost", sc(binary(), #{desc => ?DESC("event_peerhost")})}, {"peerhost", sc(binary(), #{desc => ?DESC("event_peerhost")})},
@ -283,19 +244,16 @@ fields("ctx_check_authz_complete") ->
]; ];
fields("ctx_bridge_mqtt") -> fields("ctx_bridge_mqtt") ->
[ [
{"event_type", {"event_type", event_type_sc('$bridges/mqtt:*')},
sc('$bridges/mqtt:*', #{desc => ?DESC("event_event_type"), required => true})},
{"id", sc(binary(), #{desc => ?DESC("event_id")})}, {"id", sc(binary(), #{desc => ?DESC("event_id")})},
{"payload", sc(binary(), #{desc => ?DESC("event_payload")})}, {"payload", sc(binary(), #{desc => ?DESC("event_payload")})},
{"topic", sc(binary(), #{desc => ?DESC("event_topic")})}, {"topic", sc(binary(), #{desc => ?DESC("event_topic")})},
{"server", sc(binary(), #{desc => ?DESC("event_server")})}, {"server", sc(binary(), #{desc => ?DESC("event_server")})},
{"dup", sc(binary(), #{desc => ?DESC("event_dup")})}, {"dup", sc(binary(), #{desc => ?DESC("event_dup")})},
{"retain", sc(binary(), #{desc => ?DESC("event_retain")})}, {"retain", sc(binary(), #{desc => ?DESC("event_retain")})},
{"message_received_at", {"message_received_at", publish_received_at_sc()},
sc(integer(), #{ qos()
desc => ?DESC("event_publish_received_at") ].
})}
] ++ [qos()].
qos() -> qos() ->
{"qos", sc(emqx_schema:qos(), #{desc => ?DESC("event_qos")})}. {"qos", sc(emqx_schema:qos(), #{desc => ?DESC("event_qos")})}.
@ -312,4 +270,22 @@ rule_id() ->
)}. )}.
sc(Type, Meta) -> hoconsc:mk(Type, Meta). sc(Type, Meta) -> hoconsc:mk(Type, Meta).
ref(Field) -> hoconsc:ref(?MODULE, Field). ref(Field) -> hoconsc:ref(?MODULE, Field).
event_type_sc(Event) ->
sc(Event, #{desc => ?DESC("event_event_type"), required => true}).
publish_received_at_sc() ->
sc(integer(), #{desc => ?DESC("event_publish_received_at")}).
msg_event_common_fields() ->
[
{"clientid", sc(binary(), #{desc => ?DESC("event_clientid")})},
{"username", sc(binary(), #{desc => ?DESC("event_username")})},
{"payload", sc(binary(), #{desc => ?DESC("event_payload")})},
{"peerhost", sc(binary(), #{desc => ?DESC("event_peerhost")})},
{"topic", sc(binary(), #{desc => ?DESC("event_topic")})},
{"publish_received_at", publish_received_at_sc()},
qos()
].

2
apps/emqx_rule_engine/src/emqx_rule_engine.app.src
View File

@ -2,7 +2,7 @@
{application, emqx_rule_engine, [ {application, emqx_rule_engine, [
{description, "EMQX Rule Engine"}, {description, "EMQX Rule Engine"},
% strict semver, bump manually! % strict semver, bump manually!
{vsn, "5.0.4"}, {vsn, "5.0.5"},
{modules, []}, {modules, []},
{registered, [emqx_rule_engine_sup, emqx_rule_engine]}, {registered, [emqx_rule_engine_sup, emqx_rule_engine]},
{applications, [kernel, stdlib, rulesql, getopt]}, {applications, [kernel, stdlib, rulesql, getopt]},

78
apps/emqx_rule_engine/src/emqx_rule_events.erl
View File

@ -28,6 +28,7 @@
unload/1, unload/1,
event_names/0, event_names/0,
event_name/1, event_name/1,
event_topics_enum/0,
event_topic/1, event_topic/1,
eventmsg_publish/1 eventmsg_publish/1
]). ]).
@ -78,6 +79,22 @@ event_names() ->
'delivery.dropped' 'delivery.dropped'
]. ].
%% for documentation purposes
event_topics_enum() ->
[
'$events/client_connected',
'$events/client_disconnected',
'$events/client_connack',
'$events/client_check_authz_complete',
'$events/session_subscribed',
'$events/session_unsubscribed',
'$events/message_delivered',
'$events/message_acked',
'$events/message_dropped',
'$events/delivery_dropped'
% '$events/message_publish' % not possible to use in SELECT FROM
].
reload() -> reload() ->
lists:foreach( lists:foreach(
fun(Rule) -> fun(Rule) ->
@ -88,21 +105,22 @@ reload() ->
load(Topic) -> load(Topic) ->
HookPoint = event_name(Topic), HookPoint = event_name(Topic),
HookFun = hook_fun_name(HookPoint),
emqx_hooks:put( emqx_hooks:put(
HookPoint, {?MODULE, hook_fun(HookPoint), [#{event_topic => Topic}]}, ?HP_RULE_ENGINE HookPoint, {?MODULE, HookFun, [#{event_topic => Topic}]}, ?HP_RULE_ENGINE
). ).
unload() -> unload() ->
lists:foreach( lists:foreach(
fun(HookPoint) -> fun(HookPoint) ->
emqx_hooks:del(HookPoint, {?MODULE, hook_fun(HookPoint)}) emqx_hooks:del(HookPoint, {?MODULE, hook_fun_name(HookPoint)})
end, end,
event_names() event_names()
). ).
unload(Topic) -> unload(Topic) ->
HookPoint = event_name(Topic), HookPoint = event_name(Topic),
emqx_hooks:del(HookPoint, {?MODULE, hook_fun(HookPoint)}). emqx_hooks:del(HookPoint, {?MODULE, hook_fun_name(HookPoint)}).
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% Callbacks %% Callbacks
@ -987,15 +1005,25 @@ columns_example_props_specific(unsub_props) ->
%% Helper functions %% Helper functions
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
hook_fun(<<"$bridges/", _/binary>>) -> hook_fun_name(HookPoint) ->
on_bridge_message_received; HookFun = hook_fun(HookPoint),
hook_fun(Event) -> {name, HookFunName} = erlang:fun_info(HookFun, name),
case string:split(atom_to_list(Event), ".") of HookFunName.
[Prefix, Name] ->
list_to_atom(lists:append(["on_", Prefix, "_", Name])); %% return static function references to help static code checks
[_] -> hook_fun(<<"$bridges/", _/binary>>) -> fun ?MODULE:on_bridge_message_received/2;
error(invalid_event, Event) hook_fun('client.connected') -> fun ?MODULE:on_client_connected/3;
end. hook_fun('client.disconnected') -> fun ?MODULE:on_client_disconnected/4;
hook_fun('client.connack') -> fun ?MODULE:on_client_connack/4;
hook_fun('client.check_authz_complete') -> fun ?MODULE:on_client_check_authz_complete/6;
hook_fun('session.subscribed') -> fun ?MODULE:on_session_subscribed/4;
hook_fun('session.unsubscribed') -> fun ?MODULE:on_session_unsubscribed/4;
hook_fun('message.delivered') -> fun ?MODULE:on_message_delivered/3;
hook_fun('message.acked') -> fun ?MODULE:on_message_acked/3;
hook_fun('message.dropped') -> fun ?MODULE:on_message_dropped/4;
hook_fun('delivery.dropped') -> fun ?MODULE:on_delivery_dropped/4;
hook_fun('message.publish') -> fun ?MODULE:on_message_publish/2;
hook_fun(Event) -> error({invalid_event, Event}).
reason(Reason) when is_atom(Reason) -> Reason; reason(Reason) when is_atom(Reason) -> Reason;
reason({shutdown, Reason}) when is_atom(Reason) -> Reason; reason({shutdown, Reason}) when is_atom(Reason) -> Reason;
@ -1006,19 +1034,20 @@ ntoa(undefined) -> undefined;
ntoa({IpAddr, Port}) -> iolist_to_binary([inet:ntoa(IpAddr), ":", integer_to_list(Port)]); ntoa({IpAddr, Port}) -> iolist_to_binary([inet:ntoa(IpAddr), ":", integer_to_list(Port)]);
ntoa(IpAddr) -> iolist_to_binary(inet:ntoa(IpAddr)). ntoa(IpAddr) -> iolist_to_binary(inet:ntoa(IpAddr)).
event_name(<<"$events/client_connected", _/binary>>) -> 'client.connected'; event_name(<<"$bridges/", _/binary>> = Bridge) -> Bridge;
event_name(<<"$events/client_disconnected", _/binary>>) -> 'client.disconnected'; event_name(<<"$events/client_connected">>) -> 'client.connected';
event_name(<<"$events/client_connack", _/binary>>) -> 'client.connack'; event_name(<<"$events/client_disconnected">>) -> 'client.disconnected';
event_name(<<"$events/client_check_authz_complete", _/binary>>) -> 'client.check_authz_complete'; event_name(<<"$events/client_connack">>) -> 'client.connack';
event_name(<<"$events/session_subscribed", _/binary>>) -> 'session.subscribed'; event_name(<<"$events/client_check_authz_complete">>) -> 'client.check_authz_complete';
event_name(<<"$events/session_unsubscribed", _/binary>>) -> 'session.unsubscribed'; event_name(<<"$events/session_subscribed">>) -> 'session.subscribed';
event_name(<<"$events/message_delivered", _/binary>>) -> 'message.delivered'; event_name(<<"$events/session_unsubscribed">>) -> 'session.unsubscribed';
event_name(<<"$events/message_acked", _/binary>>) -> 'message.acked'; event_name(<<"$events/message_delivered">>) -> 'message.delivered';
event_name(<<"$events/message_dropped", _/binary>>) -> 'message.dropped'; event_name(<<"$events/message_acked">>) -> 'message.acked';
event_name(<<"$events/delivery_dropped", _/binary>>) -> 'delivery.dropped'; event_name(<<"$events/message_dropped">>) -> 'message.dropped';
event_name(<<"$bridges/", _/binary>> = Topic) -> Topic; event_name(<<"$events/delivery_dropped">>) -> 'delivery.dropped';
event_name(_) -> 'message.publish'. event_name(_) -> 'message.publish'.
event_topic(<<"$bridges/", _/binary>> = Bridge) -> Bridge;
event_topic('client.connected') -> <<"$events/client_connected">>; event_topic('client.connected') -> <<"$events/client_connected">>;
event_topic('client.disconnected') -> <<"$events/client_disconnected">>; event_topic('client.disconnected') -> <<"$events/client_disconnected">>;
event_topic('client.connack') -> <<"$events/client_connack">>; event_topic('client.connack') -> <<"$events/client_connack">>;
@ -1029,8 +1058,7 @@ event_topic('message.delivered') -> <<"$events/message_delivered">>;
event_topic('message.acked') -> <<"$events/message_acked">>; event_topic('message.acked') -> <<"$events/message_acked">>;
event_topic('message.dropped') -> <<"$events/message_dropped">>; event_topic('message.dropped') -> <<"$events/message_dropped">>;
event_topic('delivery.dropped') -> <<"$events/delivery_dropped">>; event_topic('delivery.dropped') -> <<"$events/delivery_dropped">>;
event_topic('message.publish') -> <<"$events/message_publish">>; event_topic('message.publish') -> <<"$events/message_publish">>.
event_topic(<<"$bridges/", _/binary>> = Topic) -> Topic.
printable_maps(undefined) -> printable_maps(undefined) ->
#{}; #{};

46
apps/emqx_rule_engine/test/emqx_rule_events_SUITE.erl
View File

@ -8,19 +8,18 @@
all() -> emqx_common_test_helpers:all(?MODULE). all() -> emqx_common_test_helpers:all(?MODULE).
t_mod_hook_fun(_) -> t_mod_hook_fun(_) ->
Funcs = emqx_rule_events:module_info(exports), Events = emqx_rule_events:event_names(),
[ lists:foreach(
?assert(lists:keymember(emqx_rule_events:hook_fun(Event), 1, Funcs)) fun(E) ->
|| Event <- [ ?assert(is_function(emqx_rule_events:hook_fun(E)))
'client.connected', end,
'client.disconnected', Events
'session.subscribed', ),
'session.unsubscribed', ?assertEqual(
'message.acked', fun emqx_rule_events:on_bridge_message_received/2,
'message.dropped', emqx_rule_events:hook_fun(<<"$bridges/foo">>)
'message.delivered' ),
] ?assertError({invalid_event, foo}, emqx_rule_events:hook_fun(foo)).
].
t_printable_maps(_) -> t_printable_maps(_) ->
Headers = #{ Headers = #{
@ -42,3 +41,24 @@ t_printable_maps(_) ->
?assertNot(maps:is_key(redispatch_to, Converted)), ?assertNot(maps:is_key(redispatch_to, Converted)),
?assertNot(maps:is_key(shared_dispatch_ack, Converted)), ?assertNot(maps:is_key(shared_dispatch_ack, Converted)),
ok. ok.
t_event_name_topic_conversion(_) ->
Events = emqx_rule_events:event_names() -- ['message.publish'],
Topics = [atom_to_binary(A) || A <- emqx_rule_events:event_topics_enum()],
Zip = lists:zip(Events, Topics),
lists:foreach(
fun({Event, Topic}) ->
?assertEqual(Event, emqx_rule_events:event_name(Topic)),
?assertEqual(Topic, emqx_rule_events:event_topic(Event))
end,
Zip
).
t_special_events_name_topic_conversion(_) ->
Bridge = <<"$bridges/foo:bar">>,
AdHoc = <<"foo/bar">>,
NonExisting = <<"$events/message_publish">>,
?assertEqual(Bridge, emqx_rule_events:event_name(Bridge)),
?assertEqual('message.publish', emqx_rule_events:event_name(AdHoc)),
?assertEqual('message.publish', emqx_rule_events:event_name(NonExisting)),
?assertEqual(NonExisting, emqx_rule_events:event_topic('message.publish')).