diff --git a/apps/emqx_connector/src/emqx_connector_mysql.erl b/apps/emqx_connector/src/emqx_connector_mysql.erl index d6963d04e..409da4060 100644 --- a/apps/emqx_connector/src/emqx_connector_mysql.erl +++ b/apps/emqx_connector/src/emqx_connector_mysql.erl @@ -44,6 +44,16 @@ default_port => ?MYSQL_DEFAULT_PORT }). +-type prepares() :: #{atom() => binary()}. +-type params_tokens() :: #{atom() => list()}. +-type state() :: + #{ + poolname := atom(), + prepare_statement := prepares(), + auto_reconnect := boolean(), + params_tokens := params_tokens() + }. + %%===================================================================== %% Hocon schema roots() -> @@ -63,6 +73,7 @@ server(desc) -> ?DESC("server"); server(_) -> undefined. %% =================================================================== +-spec on_start(binary(), hoconsc:config()) -> {ok, state()} | {error, _}. on_start( InstId, #{ @@ -97,8 +108,8 @@ on_start( {pool_size, PoolSize} ], PoolName = emqx_plugin_libs_pool:pool_name(InstId), - Prepares = maps:get(prepare_statement, Config, #{}), - State = #{poolname => PoolName, prepare_statement => Prepares, auto_reconnect => AutoReconn}, + Prepares = parse_prepare_sql(maps:get(prepare_statement, Config, #{})), + State = maps:merge(#{poolname => PoolName, auto_reconnect => AutoReconn}, Prepares), case emqx_plugin_libs_pool:start_pool(PoolName, ?MODULE, Options ++ SslOpts) of ok -> {ok, init_prepare(State)}; {error, Reason} -> {error, Reason} @@ -111,13 +122,13 @@ on_stop(InstId, #{poolname := PoolName}) -> }), emqx_plugin_libs_pool:stop_pool(PoolName). -on_query(InstId, {Type, SQLOrKey}, AfterQuery, State) -> - on_query(InstId, {Type, SQLOrKey, [], default_timeout}, AfterQuery, State); -on_query(InstId, {Type, SQLOrKey, Params}, AfterQuery, State) -> - on_query(InstId, {Type, SQLOrKey, Params, default_timeout}, AfterQuery, State); +on_query(InstId, {TypeOrKey, SQLOrKey}, AfterQuery, State) -> + on_query(InstId, {TypeOrKey, SQLOrKey, [], default_timeout}, AfterQuery, State); +on_query(InstId, {TypeOrKey, SQLOrKey, Params}, AfterQuery, State) -> + on_query(InstId, {TypeOrKey, SQLOrKey, Params, default_timeout}, AfterQuery, State); on_query( InstId, - {Type, SQLOrKey, Params, Timeout}, + {TypeOrKey, SQLOrKey, Params, Timeout}, AfterQuery, #{poolname := PoolName, prepare_statement := Prepares} = State ) -> @@ -125,8 +136,9 @@ on_query( ?TRACE("QUERY", "mysql_connector_received", LogMeta), Worker = ecpool:get_client(PoolName), {ok, Conn} = ecpool_worker:client(Worker), - MySqlFunction = mysql_function(Type), - Result = erlang:apply(mysql, MySqlFunction, [Conn, SQLOrKey, Params, Timeout]), + MySqlFunction = mysql_function(TypeOrKey), + {SQLOrKey2, Data} = proc_sql_params(TypeOrKey, SQLOrKey, Params, State), + Result = erlang:apply(mysql, MySqlFunction, [Conn, SQLOrKey2, Data, Timeout]), case Result of {error, disconnected} -> ?SLOG( @@ -145,7 +157,7 @@ on_query( case prepare_sql(Prepares, PoolName) of ok -> %% not return result, next loop will try again - on_query(InstId, {Type, SQLOrKey, Params, Timeout}, AfterQuery, State); + on_query(InstId, {TypeOrKey, SQLOrKey, Params, Timeout}, AfterQuery, State); {error, Reason} -> ?SLOG( error, @@ -166,8 +178,13 @@ on_query( Result end. -mysql_function(sql) -> query; -mysql_function(prepared_query) -> execute. +mysql_function(sql) -> + query; +mysql_function(prepared_query) -> + execute; +%% for bridge +mysql_function(_) -> + mysql_function(prepared_query). on_get_status(_InstId, #{poolname := Pool, auto_reconnect := AutoReconn} = State) -> case emqx_plugin_libs_pool:health_check_ecpool_workers(Pool, fun ?MODULE:do_get_status/1) of @@ -287,3 +304,24 @@ prepare_sql_to_conn(Conn, [{Key, SQL} | PrepareList]) when is_pid(Conn) -> unprepare_sql_to_conn(Conn, PrepareSqlKey) -> mysql:unprepare(Conn, PrepareSqlKey). + +parse_prepare_sql(SQL) -> + parse_prepare_sql(maps:to_list(SQL), #{}, #{}). + +parse_prepare_sql([{Key, H} | T], SQL, Tokens) -> + {PrepareSQL, ParamsTokens} = emqx_plugin_libs_rule:preproc_sql(H), + parse_prepare_sql(T, SQL#{Key => PrepareSQL}, Tokens#{Key => ParamsTokens}); +parse_prepare_sql([], SQL, Tokens) -> + #{prepare_statement => SQL, params_tokens => Tokens}. + +proc_sql_params(query, SQLOrKey, Params, _State) -> + {SQLOrKey, Params}; +proc_sql_params(prepared_query, SQLOrKey, Params, _State) -> + {SQLOrKey, Params}; +proc_sql_params(TypeOrKey, SQLOrData, Params, #{params_tokens := ParamsTokens}) -> + case maps:get(TypeOrKey, ParamsTokens, undefined) of + undefined -> + {SQLOrData, Params}; + Tokens -> + {TypeOrKey, emqx_plugin_libs_rule:proc_sql(Tokens, SQLOrData)} + end. diff --git a/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.erl b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.erl index 5132ee249..a690b50d0 100644 --- a/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.erl +++ b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.erl @@ -42,7 +42,7 @@ conn_bridge_examples(Method) -> resource_type(Type) when is_binary(Type) -> resource_type(binary_to_atom(Type, utf8)); resource_type(hstreamdb) -> emqx_ee_connector_hstream; -resource_type(mysql) -> emqx_ee_connector_mysql; +resource_type(mysql) -> emqx_connector_mysql; resource_type(influxdb_udp) -> emqx_ee_connector_influxdb; resource_type(influxdb_api_v1) -> emqx_ee_connector_influxdb; resource_type(influxdb_api_v2) -> emqx_ee_connector_influxdb. diff --git a/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_mysql.erl b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_mysql.erl index 7d586d7eb..bf38da2f8 100644 --- a/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_mysql.erl +++ b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_mysql.erl @@ -50,11 +50,11 @@ values(post) -> pool_size => 8, username => <<"root">>, password => <<"public">>, - auto_reconnect => true + auto_reconnect => true, + prepare_statement => #{send_message => ?DEFAULT_SQL} }, enable => true, - direction => egress, - sql => ?DEFAULT_SQL + direction => egress }; values(put) -> values(post). @@ -69,27 +69,29 @@ fields("config") -> [ {enable, mk(boolean(), #{desc => ?DESC("config_enable"), default => true})}, {direction, mk(egress, #{desc => ?DESC("config_direction"), default => egress})}, - {sql, mk(binary(), #{default => ?DEFAULT_SQL, desc => ?DESC("sql")})}, - {connector, field(connector)} + {connector, + mk( + ref(?MODULE, connector), + #{ + required => true, + desc => ?DESC("desc_connector") + } + )} ]; fields("post") -> [type_field(), name_field() | fields("config")]; fields("put") -> fields("config"); fields("get") -> - emqx_bridge_schema:metrics_status_fields() ++ fields("post"). - -field(connector) -> - mk( - ref(emqx_ee_connector_mysql, config), - #{ - required => true, - desc => ?DESC("desc_connector") - } - ). + emqx_bridge_schema:metrics_status_fields() ++ fields("post"); +fields(connector) -> + (emqx_connector_mysql:fields(config) -- + emqx_connector_schema_lib:prepare_statement_fields()) ++ prepare_statement_fields(). desc("config") -> ?DESC("desc_config"); +desc(connector) -> + ?DESC("desc_connector"); desc(Method) when Method =:= "get"; Method =:= "put"; Method =:= "post" -> ["Configuration for MySQL using `", string:to_upper(Method), "` method."]; desc(_) -> @@ -102,3 +104,13 @@ type_field() -> name_field() -> {name, mk(binary(), #{required => true, desc => ?DESC("desc_name")})}. + +prepare_statement_fields() -> + [ + {prepare_statement, + mk(map(), #{ + desc => ?DESC(emqx_connector_schema_lib, prepare_statement), + default => #{<<"send_message">> => ?DEFAULT_SQL}, + example => #{<<"send_message">> => ?DEFAULT_SQL} + })} + ]. diff --git a/lib-ee/emqx_ee_connector/i18n/emqx_ee_connector_mysql.conf b/lib-ee/emqx_ee_connector/i18n/emqx_ee_connector_mysql.conf deleted file mode 100644 index 5d7c4848a..000000000 --- a/lib-ee/emqx_ee_connector/i18n/emqx_ee_connector_mysql.conf +++ /dev/null @@ -1,23 +0,0 @@ -emqx_ee_connector_mysql { - type { - desc { - en: "The Connector Type." - zh: "连接器类型。" - } - label: { - en: "Connector Type" - zh: "连接器类型" - } - } - - name { - desc { - en: "Connector name, used as a human-readable description of the connector." - zh: "连接器名称，人类可读的连接器描述。" - } - label: { - en: "Connector Name" - zh: "连接器名称" - } - } -} diff --git a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_mysql.erl b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_mysql.erl deleted file mode 100644 index c4efff91c..000000000 --- a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_mysql.erl +++ /dev/null @@ -1,66 +0,0 @@ -%%-------------------------------------------------------------------- -%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved. -%%-------------------------------------------------------------------- --module(emqx_ee_connector_mysql). - --include_lib("hocon/include/hoconsc.hrl"). --include_lib("typerefl/include/types.hrl"). --include_lib("emqx/include/logger.hrl"). - --import(hoconsc, [mk/2, enum/1]). - --behaviour(emqx_resource). - -%% callbacks of behaviour emqx_resource --export([ - on_start/2, - on_stop/2, - on_query/4, - on_get_status/2 -]). - --export([ - roots/0, - fields/1 -]). - --define(SEND_MSG_KEY, send_message). - -%% ------------------------------------------------------------------------------------------------- -%% resource callback - -on_start(InstId, #{egress := #{sql := SQL}} = Config) -> - {PrepareSQL, ParamsTokens} = emqx_plugin_libs_rule:preproc_sql(SQL, '?'), - {ok, State} = emqx_connector_mysql:on_start(InstId, Config#{ - prepare_statement => #{?SEND_MSG_KEY => PrepareSQL} - }), - {ok, State#{'ParamsTokens' => ParamsTokens}}. - -on_stop(InstId, State) -> - emqx_connector_mysql:on_stop(InstId, State). - -on_query( - InstId, - {?SEND_MSG_KEY, Msg}, - AfterQuery, - #{'ParamsTokens' := ParamsTokens} = State -) -> - Data = emqx_plugin_libs_rule:proc_sql(ParamsTokens, Msg), - emqx_connector_mysql:on_query( - InstId, {prepared_query, ?SEND_MSG_KEY, Data}, AfterQuery, State - ). - -on_get_status(InstId, State) -> - emqx_connector_mysql:on_get_status(InstId, State). - -%% ------------------------------------------------------------------------------------------------- -%% schema - -roots() -> - fields(config). - -fields(config) -> - emqx_connector_mysql:fields(config) -- emqx_connector_schema_lib:prepare_statement_fields(). - -%% ------------------------------------------------------------------------------------------------- -%% internal functions