fix(bridge): remove emqx_ee_connector_mysql

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.

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.

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}
+            })}
+    ].

lib-ee/emqx_ee_connector/i18n/emqx_ee_connector_mysql.conf

@@ -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: "连接器名称"
-            }
-    }
-}

lib-ee/emqx_ee_connector/src/emqx_ee_connector_mysql.erl

@@ -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