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/i18n/emqx_ee_bridge_mysql.conf b/lib-ee/emqx_ee_bridge/i18n/emqx_ee_bridge_mysql.conf new file mode 100644 index 000000000..9e3796487 --- /dev/null +++ b/lib-ee/emqx_ee_bridge/i18n/emqx_ee_bridge_mysql.conf @@ -0,0 +1,75 @@ +emqx_ee_bridge_mysql { + sql { + desc { + en: """SQL Template""" + zh: """SQL 模板""" + } + label { + en: "SQL Template" + zh: "SQL 模板" + } + } + config_enable { + desc { + en: """Enable or disable this bridge""" + zh: """启用/禁用桥接""" + } + label { + en: "Enable Or Disable Bridge" + zh: "启用/禁用桥接" + } + } + config_direction { + desc { + en: """The direction of this bridge, MUST be 'egress'""" + zh: """桥接的方向, 必须是 egress""" + } + label { + en: "Bridge Direction" + zh: "桥接方向" + } + } + + desc_config { + desc { + en: """Configuration for an HStreamDB bridge.""" + zh: """HStreamDB 桥接配置""" + } + label: { + en: "HStreamDB Bridge Configuration" + zh: "HStreamDB 桥接配置" + } + } + + desc_type { + desc { + en: """The Bridge Type""" + zh: """Bridge 类型""" + } + label { + en: "Bridge Type" + zh: "桥接类型" + } + } + + desc_name { + desc { + en: """Bridge name, used as a human-readable description of the bridge.""" + zh: """桥接名字,可读描述""" + } + label { + en: "Bridge Name" + zh: "桥接名字" + } + } + desc_connector { + desc { + en: """Generic configuration for the connector.""" + zh: """连接器的通用配置。""" + } + label: { + en: "Connector Generic Configuration" + zh: "连接器通用配置。" + } + } +} 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 881609bfc..a690b50d0 100644 --- a/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.erl +++ b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.erl @@ -15,6 +15,7 @@ api_schemas(Method) -> [ ref(emqx_ee_bridge_hstream, Method), + ref(emqx_ee_bridge_mysql, Method), ref(emqx_ee_bridge_influxdb, Method ++ "_udp"), ref(emqx_ee_bridge_influxdb, Method ++ "_api_v1"), ref(emqx_ee_bridge_influxdb, Method ++ "_api_v2") @@ -23,7 +24,8 @@ api_schemas(Method) -> schema_modules() -> [ emqx_ee_bridge_hstream, - emqx_ee_bridge_influxdb + emqx_ee_bridge_influxdb, + emqx_ee_bridge_mysql ]. conn_bridge_examples(Method) -> @@ -40,6 +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_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. @@ -50,6 +53,11 @@ fields(bridges) -> mk( hoconsc:map(name, ref(emqx_ee_bridge_hstream, "config")), #{desc => <<"EMQX Enterprise Config">>} + )}, + {mysql, + mk( + hoconsc:map(name, ref(emqx_ee_bridge_mysql, "config")), + #{desc => <<"EMQX Enterprise Config">>} )} ] ++ fields(influxdb); fields(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 new file mode 100644 index 000000000..bf38da2f8 --- /dev/null +++ b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_mysql.erl @@ -0,0 +1,116 @@ +%%-------------------------------------------------------------------- +%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved. +%%-------------------------------------------------------------------- +-module(emqx_ee_bridge_mysql). + +-include_lib("typerefl/include/types.hrl"). +-include_lib("hocon/include/hoconsc.hrl"). +-include("emqx_ee_bridge.hrl"). + +-import(hoconsc, [mk/2, enum/1, ref/2]). + +-export([ + conn_bridge_examples/1 +]). + +-export([ + namespace/0, + roots/0, + fields/1, + desc/1 +]). + +-define(DEFAULT_SQL, << + "insert into t_mqtt_msg(msgid, topic, qos, payload, arrived) " + "values (${id}, ${topic}, ${qos}, ${payload}, FROM_UNIXTIME(${timestamp}/1000))" +>>). + +%% ------------------------------------------------------------------------------------------------- +%% api + +conn_bridge_examples(Method) -> + [ + #{ + <<"mysql">> => #{ + summary => <<"MySQL Bridge">>, + value => values(Method) + } + } + ]. + +values(get) -> + maps:merge(values(post), ?METRICS_EXAMPLE); +values(post) -> + #{ + type => mysql, + name => <<"mysql">>, + connector => #{ + server => <<"127.0.0.1:3306">>, + database => <<"test">>, + pool_size => 8, + username => <<"root">>, + password => <<"public">>, + auto_reconnect => true, + prepare_statement => #{send_message => ?DEFAULT_SQL} + }, + enable => true, + direction => egress + }; +values(put) -> + values(post). + +%% ------------------------------------------------------------------------------------------------- +%% Hocon Schema Definitions +namespace() -> "bridge". + +roots() -> []. + +fields("config") -> + [ + {enable, mk(boolean(), #{desc => ?DESC("config_enable"), default => true})}, + {direction, mk(egress, #{desc => ?DESC("config_direction"), default => egress})}, + {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"); +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(_) -> + undefined. + +%% ------------------------------------------------------------------------------------------------- +%% internal +type_field() -> + {type, mk(enum([mysql]), #{required => true, desc => ?DESC("desc_type")})}. + +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} + })} + ].