yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
emqx/apps/emqx_plugin_libs/src/emqx_plugin_libs_rule.erl

273 lines
8.3 KiB
Erlang
Raw Permalink Blame History

%%--------------------------------------------------------------------
%% Copyright (c) 2021 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%% http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_plugin_libs_rule).
-elvis([{elvis_style, god_modules, disable}]).
%% preprocess and process template string with place holders
-export([ preproc_tmpl/1
, proc_tmpl/2
, proc_tmpl/3
, preproc_cmd/1
, proc_cmd/2
, proc_cmd/3
, preproc_sql/1
, preproc_sql/2
, proc_sql/2
, proc_sql_param_str/2
, proc_cql_param_str/2
]).
%% type converting
-export([ str/1
, bin/1
, bool/1
, int/1
, float/1
, map/1
, utf8_bin/1
, utf8_str/1
, number_to_binary/1
, atom_key/1
, unsafe_atom_key/1
]).
%% connectivity check
-export([ http_connectivity/1
, http_connectivity/2
, tcp_connectivity/2
, tcp_connectivity/3
]).
-export([ now_ms/0
, can_topic_match_oneof/2
]).
-export([cluster_call/3]).
-compile({no_auto_import,
[ float/1
]}).
-define(EX_PLACE_HOLDER, "(\\$\\{[a-zA-Z0-9\\._]+\\})").
-define(EX_WITHE_CHARS, "\\s"). %% Space and CRLF
-type(uri_string() :: iodata()).
-type(tmpl_token() :: list({var, binary()} | {str, binary()})).
-type(tmpl_cmd() :: list(tmpl_token())).
-type(prepare_statement_key() :: binary()).
%% preprocess template string with place holders
-spec(preproc_tmpl(binary()) -> tmpl_token()).
preproc_tmpl(Str) ->
emqx_placeholder:preproc_tmpl(Str).
-spec(proc_tmpl(tmpl_token(), map()) -> binary()).
proc_tmpl(Tokens, Data) ->
emqx_placeholder:proc_tmpl(Tokens, Data).
-spec(proc_tmpl(tmpl_token(), map(), map()) -> binary() | list()).
proc_tmpl(Tokens, Data, Opts) ->
emqx_placeholder:proc_tmpl(Tokens, Data, Opts).
-spec(preproc_cmd(binary()) -> tmpl_cmd()).
preproc_cmd(Str) ->
emqx_placeholder:preproc_cmd(Str).
-spec(proc_cmd([tmpl_token()], map()) -> binary() | list()).
proc_cmd(Tokens, Data) ->
emqx_placeholder:proc_cmd(Tokens, Data).
-spec(proc_cmd([tmpl_token()], map(), map()) -> list()).
proc_cmd(Tokens, Data, Opts) ->
emqx_placeholder:proc_cmd(Tokens, Data, Opts).
%% preprocess SQL with place holders
-spec(preproc_sql(Sql::binary()) -> {prepare_statement_key(), tmpl_token()}).
preproc_sql(Sql) ->
emqx_placeholder:preproc_sql(Sql).
-spec(preproc_sql(Sql::binary(), ReplaceWith :: '?' | '$n')
-> {prepare_statement_key(), tmpl_token()}).
preproc_sql(Sql, ReplaceWith) ->
emqx_placeholder:preproc_sql(Sql, ReplaceWith).
-spec(proc_sql(tmpl_token(), map()) -> list()).
proc_sql(Tokens, Data) ->
emqx_placeholder:proc_sql(Tokens, Data).
-spec(proc_sql_param_str(tmpl_token(), map()) -> binary()).
proc_sql_param_str(Tokens, Data) ->
emqx_placeholder:proc_sql_param_str(Tokens, Data).
-spec(proc_cql_param_str(tmpl_token(), map()) -> binary()).
proc_cql_param_str(Tokens, Data) ->
emqx_placeholder:proc_cql_param_str(Tokens, Data).
unsafe_atom_key(Key) when is_atom(Key) ->
Key;
unsafe_atom_key(Key) when is_binary(Key) ->
binary_to_atom(Key, utf8);
unsafe_atom_key(Keys = [_Key | _]) ->
[unsafe_atom_key(SubKey) || SubKey <- Keys];
unsafe_atom_key(Key) ->
error({invalid_key, Key}).
atom_key(Key) when is_atom(Key) ->
Key;
atom_key(Key) when is_binary(Key) ->
try binary_to_existing_atom(Key, utf8)
catch error:badarg -> error({invalid_key, Key})
end;
atom_key(Keys = [_Key | _]) -> %% nested keys
[atom_key(SubKey) || SubKey <- Keys];
atom_key(Key) ->
error({invalid_key, Key}).
-spec(http_connectivity(uri_string()) -> ok | {error, Reason :: term()}).
http_connectivity(Url) ->
http_connectivity(Url, 3000).
-spec(http_connectivity(uri_string(), integer()) -> ok | {error, Reason :: term()}).
http_connectivity(Url, Timeout) ->
case emqx_http_lib:uri_parse(Url) of
{ok, #{host := Host, port := Port}} ->
tcp_connectivity(Host, Port, Timeout);
{error, Reason} ->
{error, Reason}
end.
-spec tcp_connectivity(Host :: inet:socket_address() | inet:hostname(),
Port :: inet:port_number())
-> ok | {error, Reason :: term()}.
tcp_connectivity(Host, Port) ->
tcp_connectivity(Host, Port, 3000).
-spec(tcp_connectivity(Host :: inet:socket_address() | inet:hostname(),
Port :: inet:port_number(),
Timeout :: integer())
-> ok | {error, Reason :: term()}).
tcp_connectivity(Host, Port, Timeout) ->
case gen_tcp:connect(Host, Port, emqx_misc:ipv6_probe([]), Timeout) of
{ok, Sock} -> gen_tcp:close(Sock), ok;
{error, Reason} -> {error, Reason}
end.
str(Bin) when is_binary(Bin) -> binary_to_list(Bin);
str(Num) when is_number(Num) -> number_to_list(Num);
str(Atom) when is_atom(Atom) -> atom_to_list(Atom);
str(Map) when is_map(Map) -> binary_to_list(emqx_json:encode(Map));
str(List) when is_list(List) ->
case io_lib:printable_list(List) of
true -> List;
false -> binary_to_list(emqx_json:encode(List))
end;
str(Data) -> error({invalid_str, Data}).
utf8_bin(Str) when is_binary(Str); is_list(Str) ->
unicode:characters_to_binary(Str);
utf8_bin(Str) ->
unicode:characters_to_binary(bin(Str)).
utf8_str(Str) when is_binary(Str); is_list(Str) ->
unicode:characters_to_list(Str);
utf8_str(Str) ->
unicode:characters_to_list(str(Str)).
bin(Bin) when is_binary(Bin) -> Bin;
bin(Num) when is_number(Num) -> number_to_binary(Num);
bin(Atom) when is_atom(Atom) -> atom_to_binary(Atom, utf8);
bin(Map) when is_map(Map) -> emqx_json:encode(Map);
bin(List) when is_list(List) ->
case io_lib:printable_list(List) of
true -> list_to_binary(List);
false -> emqx_json:encode(List)
end;
bin(Data) -> error({invalid_bin, Data}).
int(List) when is_list(List) ->
try list_to_integer(List)
catch error:badarg ->
int(list_to_float(List))
end;
int(Bin) when is_binary(Bin) ->
try binary_to_integer(Bin)
catch error:badarg ->
int(binary_to_float(Bin))
end;
int(Int) when is_integer(Int) -> Int;
int(Float) when is_float(Float) -> erlang:floor(Float);
int(true) -> 1;
int(false) -> 0;
int(Data) -> error({invalid_number, Data}).
float(List) when is_list(List) ->
try list_to_float(List)
catch error:badarg ->
float(list_to_integer(List))
end;
float(Bin) when is_binary(Bin) ->
try binary_to_float(Bin)
catch error:badarg ->
float(binary_to_integer(Bin))
end;
float(Num) when is_number(Num) -> erlang:float(Num);
float(Data) -> error({invalid_number, Data}).
map(Bin) when is_binary(Bin) ->
case emqx_json:decode(Bin, [return_maps]) of
Map = #{} -> Map;
_ -> error({invalid_map, Bin})
end;
map(List) when is_list(List) -> maps:from_list(List);
map(Map) when is_map(Map) -> Map;
map(Data) -> error({invalid_map, Data}).
bool(Bool) when Bool == true;
Bool == <<"true">>;
Bool == 1 -> true;
bool(Bool) when Bool == false;
Bool == <<"false">>;
Bool == 0 -> false;
bool(Bool) -> error({invalid_boolean, Bool}).
number_to_binary(Int) when is_integer(Int) ->
integer_to_binary(Int);
number_to_binary(Float) when is_float(Float) ->
float_to_binary(Float, [{decimals, 10}, compact]).
number_to_list(Int) when is_integer(Int) ->
integer_to_list(Int);
number_to_list(Float) when is_float(Float) ->
float_to_list(Float, [{decimals, 10}, compact]).
now_ms() ->
erlang:system_time(millisecond).
can_topic_match_oneof(Topic, Filters) ->
lists:any(fun(Fltr) ->
emqx_topic:match(Topic, Fltr)
end, Filters).
cluster_call(Module, Func, Args) ->
{ok, _TnxId, Result} = emqx_cluster_rpc:multicall(Module, Func, Args),
Result.
Reference in New Issue View Git Blame Copy Permalink