chore(connector): update schema file

Signed-off-by: zhanghongtong <rory-z@outlook.com>

apps/emqx_authz/src/emqx_authz.erl

@@ -64,15 +64,10 @@ create_resource(#{type := DB,
                   config := Config
                  } = Rule) ->
     ResourceID = iolist_to_binary([io_lib:format("~s_~s",[?APP, DB]), "_", integer_to_list(erlang:system_time())]),
-    NConfig = case DB of
-                  redis -> #{config => Config };
-                  mongo -> #{config => Config };
-                  _ -> Config
-              end,
-    case emqx_resource:check_and_create(
+    case emqx_resource:create(
             ResourceID,
             list_to_existing_atom(io_lib:format("~s_~s",[emqx_connector, DB])),
-            NConfig)
+            Config)
     of
         {ok, _} ->
             Rule#{resource_id => ResourceID};

apps/emqx_authz/src/emqx_authz_schema.erl

@@ -16,30 +16,20 @@ structs() -> ["emqx_authz"].
 fields("emqx_authz") ->
     [ {rules, rules()}
     ];
-fields(mongo_connector) ->
-    [ {principal, principal()}
-    , {type, #{type => hoconsc:enum([mongo])}}
-    , {config, #{type => map()}}
-    , {collection, #{type => atom()}}
+fields(mongo) ->
+    connector_fields(mongo) ++
+    [ {collection, #{type => atom()}}
     , {find, #{type => map()}}
     ];
-fields(redis_connector) ->
-    [ {principal, principal()}
-    , {type, #{type => hoconsc:enum([redis])}}
-    , {config, #{type => hoconsc:union(
-                         [ hoconsc:ref(emqx_connector_redis, cluster)
-                         , hoconsc:ref(emqx_connector_redis, sentinel)
-                         , hoconsc:ref(emqx_connector_redis, single)
-                         ])}
-      }
-    , {cmd, query()}
-    ];
-fields(sql_connector) ->
-    [ {principal, principal() }
-    , {type, #{type => hoconsc:enum([mysql, pgsql])}}
-    , {config, #{type => map()}}
-    , {sql, query()}
-    ];
+fields(redis) ->
+    connector_fields(redis) ++
+    [ {cmd, query()} ];
+fields(mysql) ->
+    connector_fields(mysql) ++
+    [ {sql, query()} ];
+fields(pgsql) ->
+    connector_fields(pgsql) ++
+    [ {sql, query()} ];
 fields(simple_rule) ->
     [ {permission,   #{type => permission()}}
     , {action,   #{type => action()}}
@@ -88,9 +78,10 @@ union_array(Item) when is_list(Item) ->
 rules() -> 
     #{type => union_array(
                 [ hoconsc:ref(?MODULE, simple_rule)
-                , hoconsc:ref(?MODULE, sql_connector)
-                , hoconsc:ref(?MODULE, redis_connector)
-                , hoconsc:ref(?MODULE, mongo_connector)
+                , hoconsc:ref(?MODULE, mysql)
+                , hoconsc:ref(?MODULE, pgsql)
+                , hoconsc:ref(?MODULE, redis)
+                , hoconsc:ref(?MODULE, mongo)
                 ])
     }.
 
@@ -115,3 +106,9 @@ query() ->
                          end
                        end
      }.
+
+connector_fields(DB) ->
+    Mod = list_to_existing_atom(io_lib:format("~s_~s",[emqx_connector, DB])),
+    [ {principal, principal()}
+    , {type, #{type => DB}}
+    ] ++ Mod:fields("").

apps/emqx_authz/test/emqx_authz_mongo_SUITE.erl

@@ -30,7 +30,7 @@ groups() ->
 
 init_per_suite(Config) ->
     meck:new(emqx_resource, [non_strict, passthrough, no_history, no_link]),
-    meck:expect(emqx_resource, check_and_create, fun(_, _, _) -> {ok, meck_data} end ),
+    meck:expect(emqx_resource, create, fun(_, _, _) -> {ok, meck_data} end ),
     ok = emqx_ct_helpers:start_apps([emqx_authz], fun set_special_configs/1),
     Config.
 

apps/emqx_authz/test/emqx_authz_mysql_SUITE.erl

@@ -30,7 +30,7 @@ groups() ->
 
 init_per_suite(Config) ->
     meck:new(emqx_resource, [non_strict, passthrough, no_history, no_link]),
-    meck:expect(emqx_resource, check_and_create, fun(_, _, _) -> {ok, meck_data} end ),
+    meck:expect(emqx_resource, create, fun(_, _, _) -> {ok, meck_data} end ),
     ok = emqx_ct_helpers:start_apps([emqx_authz], fun set_special_configs/1),
     Config.
 

apps/emqx_authz/test/emqx_authz_pgsql_SUITE.erl

@@ -30,7 +30,7 @@ groups() ->
 
 init_per_suite(Config) ->
     meck:new(emqx_resource, [non_strict, passthrough, no_history, no_link]),
-    meck:expect(emqx_resource, check_and_create, fun(_, _, _) -> {ok, meck_data} end ),
+    meck:expect(emqx_resource, create, fun(_, _, _) -> {ok, meck_data} end ),
     ok = emqx_ct_helpers:start_apps([emqx_authz], fun set_special_configs/1),
     Config.
 

apps/emqx_authz/test/emqx_authz_redis_SUITE.erl

@@ -30,7 +30,7 @@ groups() ->
 
 init_per_suite(Config) ->
     meck:new(emqx_resource, [non_strict, passthrough, no_history, no_link]),
-    meck:expect(emqx_resource, check_and_create, fun(_, _, _) -> {ok, meck_data} end ),
+    meck:expect(emqx_resource, create, fun(_, _, _) -> {ok, meck_data} end ),
     ok = emqx_ct_helpers:start_apps([emqx_authz], fun set_special_configs/1),
     Config.
 

apps/emqx_connector/src/emqx_connector_mongo.erl

@@ -88,24 +88,24 @@ on_jsonify(Config) ->
     Config.
 
 %% ===================================================================
-on_start(InstId, #{config := #{server := Server,
-                               mongo_type := single} = Config}) ->
+on_start(InstId, Config = #{server := Server,
+                            mongo_type := single}) ->
     logger:info("starting mongodb connector: ~p, config: ~p", [InstId, Config]),
     Opts = [{type, single},
             {hosts, [Server]}
             ],
     do_start(InstId, Opts, Config);
 
-on_start(InstId, #{config := #{servers := Servers,
-                               mongo_type := rs,
-                               replicaset_name := RsName} = Config}) ->
+on_start(InstId, Config = #{servers := Servers,
+                            mongo_type := rs,
+                            replicaset_name := RsName}) ->
     logger:info("starting mongodb connector: ~p, config: ~p", [InstId, Config]),
     Opts = [{type,  {rs, RsName}},
             {hosts, Servers}],
     do_start(InstId, Opts, Config);
 
-on_start(InstId, #{config := #{servers := Servers,
-                               mongo_type := sharded} = Config}) ->
+on_start(InstId, Config = #{servers := Servers,
+                            mongo_type := sharded}) ->
     logger:info("starting mongodb connector: ~p, config: ~p", [InstId, Config]),
     Opts = [{type, sharded},
             {hosts, Servers}

apps/emqx_connector/src/emqx_connector_mysql.erl

@@ -37,6 +37,9 @@
 structs() -> [""].
 
 fields("") ->
+    [{config, #{type => hoconsc:ref(?MODULE, config)}}];
+
+fields(config) ->
     emqx_connector_schema_lib:relational_db_fields() ++
     emqx_connector_schema_lib:ssl_fields().
 

apps/emqx_connector/src/emqx_connector_pgsql.erl

@@ -38,6 +38,9 @@
 structs() -> [""].
 
 fields("") ->
+    [{config, #{type => hoconsc:ref(?MODULE, config)}}];
+
+fields(config) ->
     emqx_connector_schema_lib:relational_db_fields() ++
     emqx_connector_schema_lib:ssl_fields().
 

apps/emqx_connector/src/emqx_connector_redis.erl

@@ -78,11 +78,11 @@ on_jsonify(Config) ->
     Config.
 
 %% ===================================================================
-on_start(InstId, #{config :=#{redis_type := Type,
-                              database := Database,
-                              pool_size := PoolSize,
-                              auto_reconnect := AutoReconn,
-                              ssl := SSL } = Config}) ->
+on_start(InstId, #{redis_type := Type,
+                   database := Database,
+                   pool_size := PoolSize,
+                   auto_reconnect := AutoReconn,
+                   ssl := SSL } = Config) ->
     logger:info("starting redis connector: ~p, config: ~p", [InstId, Config]),
     Servers = case Type of
                 single -> [{servers, [maps:get(server, Config)]}];

apps/emqx_connector/src/emqx_connector_schema_lib.erl

@@ -86,10 +86,12 @@ relational_db_fields() ->
     ].
 
 server(type) -> emqx_schema:ip_port();
+server(nullable) -> false;
 server(validator) -> [?REQUIRED("the field 'server' is required")];
 server(_) -> undefined.
 
 database(type) -> binary();
+database(nullable) -> false;
 database(validator) -> [?REQUIRED("the field 'database' is required")];
 database(_) -> undefined.
 

apps/emqx_data_bridge/src/emqx_data_bridge_schema.erl

@@ -5,11 +5,6 @@
 %%======================================================================================
 %% Hocon Schema Definitions
 
--define(BRIDGE_FIELDS(T),
-    [{name, hoconsc:t(typerefl:binary())},
-     {type, hoconsc:t(typerefl:atom(T))},
-     {config, hoconsc:t(hoconsc:ref(list_to_atom("emqx_connector_"++atom_to_list(T)), ""))}]).
-
 -define(TYPES, [mysql, pgsql, mongo, redis, ldap]).
 -define(BRIDGES, [hoconsc:ref(?MODULE, T) || T <- ?TYPES]).
 
@@ -19,8 +14,13 @@ fields("emqx_data_bridge") ->
     [{bridges, #{type => hoconsc:array(hoconsc:union(?BRIDGES)),
                  default => []}}];
 
-fields(mysql) -> ?BRIDGE_FIELDS(mysql);
-fields(pgsql) -> ?BRIDGE_FIELDS(pgsql);
-fields(mongo) -> ?BRIDGE_FIELDS(mongo);
-fields(redis) -> ?BRIDGE_FIELDS(redis);
-fields(ldap) -> ?BRIDGE_FIELDS(ldap).
+fields(mysql) -> connector_fields(mysql);
+fields(pgsql) -> connector_fields(pgsql);
+fields(mongo) -> connector_fields(mongo);
+fields(redis) -> connector_fields(redis);
+fields(ldap)  -> connector_fields(ldap).
+
+connector_fields(DB) ->
+    Mod = list_to_existing_atom(io_lib:format("~s_~s",[emqx_connector, DB])),
+    [{name, hoconsc:t(typerefl:binary())},
+     {type, #{type => DB}}] ++ Mod:fields("").

apps/emqx_resource/src/emqx_resource_validator.erl

@@ -39,7 +39,7 @@ enum(Items) ->
     end.
 
 required(ErrMsg) ->
-    fun(undefined) -> {error, ErrMsg};
+    fun(<<>>) -> {error, ErrMsg};
        (_) -> ok
     end.