refactor(schema registry): move config logic to dedicated module

apps/emqx_schema_registry/src/emqx_schema_registry.erl

@@ -4,8 +4,6 @@
 -module(emqx_schema_registry).
 
 -behaviour(gen_server).
--behaviour(emqx_config_handler).
--behaviour(emqx_config_backup).
 
 -include("emqx_schema_registry.hrl").
 -include_lib("emqx/include/logger.hrl").
@@ -31,12 +29,11 @@
     terminate/2
 ]).
 
-%% `emqx_config_handler' API
--export([post_config_update/5]).
-
-%% Data backup
+%% Internal exports for `emqx_schema_registry_config'
 -export([
-    import_config/1
+    async_delete_serdes/1,
+    ensure_serde_absent/1,
+    build_serdes/1
 ]).
 
 %% for testing
@@ -131,85 +128,6 @@ delete_schema(Name) ->
 list_schemas() ->
     emqx_config:get([?CONF_KEY_ROOT, schemas], #{}).
 
-%%-------------------------------------------------------------------------------------------------
-%% `emqx_config_handler' API
-%%-------------------------------------------------------------------------------------------------
-%% remove
-post_config_update(
-    [?CONF_KEY_ROOT, schemas, Name],
-    '$remove',
-    _NewSchemas,
-    _OldSchemas,
-    _AppEnvs
-) ->
-    async_delete_serdes([Name]),
-    ok;
-%% add or update
-post_config_update(
-    [?CONF_KEY_ROOT, schemas, NewName],
-    _Cmd,
-    NewSchema,
-    OldSchema,
-    _AppEnvs
-) ->
-    case OldSchema of
-        undefined ->
-            ok;
-        _ ->
-            ensure_serde_absent(NewName)
-    end,
-    case build_serdes([{NewName, NewSchema}]) of
-        ok ->
-            {ok, #{NewName => NewSchema}};
-        {error, Reason, SerdesToRollback} ->
-            lists:foreach(fun ensure_serde_absent/1, SerdesToRollback),
-            {error, Reason}
-    end;
-post_config_update(?CONF_KEY_PATH, _Cmd, NewConf = #{schemas := NewSchemas}, OldConf, _AppEnvs) ->
-    OldSchemas = maps:get(schemas, OldConf, #{}),
-    #{
-        added := Added,
-        changed := Changed0,
-        removed := Removed
-    } = emqx_utils_maps:diff_maps(NewSchemas, OldSchemas),
-    Changed = maps:map(fun(_N, {_Old, New}) -> New end, Changed0),
-    RemovedNames = maps:keys(Removed),
-    case RemovedNames of
-        [] ->
-            ok;
-        _ ->
-            async_delete_serdes(RemovedNames)
-    end,
-    SchemasToBuild = maps:to_list(maps:merge(Changed, Added)),
-    ok = lists:foreach(fun ensure_serde_absent/1, [N || {N, _} <- SchemasToBuild]),
-    case build_serdes(SchemasToBuild) of
-        ok ->
-            {ok, NewConf};
-        {error, Reason, SerdesToRollback} ->
-            lists:foreach(fun ensure_serde_absent/1, SerdesToRollback),
-            {error, Reason}
-    end;
-post_config_update(_Path, _Cmd, NewConf, _OldConf, _AppEnvs) ->
-    {ok, NewConf}.
-
-%%-------------------------------------------------------------------------------------------------
-%% Data backup
-%%-------------------------------------------------------------------------------------------------
-
-import_config(#{<<"schema_registry">> := #{<<"schemas">> := Schemas} = SchemaRegConf}) ->
-    OldSchemas = emqx:get_raw_config([?CONF_KEY_ROOT, schemas], #{}),
-    SchemaRegConf1 = SchemaRegConf#{<<"schemas">> => maps:merge(OldSchemas, Schemas)},
-    case emqx_conf:update(?CONF_KEY_PATH, SchemaRegConf1, #{override_to => cluster}) of
-        {ok, #{raw_config := #{<<"schemas">> := NewRawSchemas}}} ->
-            Changed = maps:get(changed, emqx_utils_maps:diff_maps(NewRawSchemas, OldSchemas)),
-            ChangedPaths = [[?CONF_KEY_ROOT, schemas, Name] || Name <- maps:keys(Changed)],
-            {ok, #{root_key => ?CONF_KEY_ROOT, changed => ChangedPaths}};
-        Error ->
-            {error, #{root_key => ?CONF_KEY_ROOT, reason => Error}}
-    end;
-import_config(_RawConf) ->
-    {ok, #{root_key => ?CONF_KEY_ROOT, changed => []}}.
-
 %%-------------------------------------------------------------------------------------------------
 %% `gen_server' API
 %%-------------------------------------------------------------------------------------------------

apps/emqx_schema_registry/src/emqx_schema_registry_app.erl

@@ -14,13 +14,9 @@ start(_StartType, _StartArgs) ->
     %% and encode functions called from the rule engine SQL like language
     ok = emqx_rule_engine:set_extra_functions_module(emqx_schema_registry_serde),
     ok = mria_rlog:wait_for_shards([?SCHEMA_REGISTRY_SHARD], infinity),
-    %% HTTP API handler
-    emqx_conf:add_handler([?CONF_KEY_ROOT, schemas, '?'], emqx_schema_registry),
-    %% Conf load / data import handler
-    emqx_conf:add_handler(?CONF_KEY_PATH, emqx_schema_registry),
+    ok = emqx_schema_registry_config:add_handlers(),
     emqx_schema_registry_sup:start_link().
 
 stop(_State) ->
-    emqx_conf:remove_handler([?CONF_KEY_ROOT, schemas, '?']),
-    emqx_conf:remove_handler(?CONF_KEY_PATH),
+    ok = emqx_schema_registry_config:remove_handlers(),
     ok.

apps/emqx_schema_registry/src/emqx_schema_registry_config.erl

@@ -0,0 +1,130 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2024 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+-module(emqx_schema_registry_config).
+
+-include("emqx_schema_registry.hrl").
+
+%% API
+-export([
+    add_handlers/0,
+    remove_handlers/0
+]).
+
+%% `emqx_config_handler' API
+-export([post_config_update/5]).
+
+%% `emqx_config_backup' API
+-behaviour(emqx_config_backup).
+-export([import_config/1]).
+
+%%------------------------------------------------------------------------------
+%% Type declarations
+%%------------------------------------------------------------------------------
+
+-define(SCHEMA_CONF_PATH(NAME), [?CONF_KEY_ROOT, schemas, NAME]).
+
+%%------------------------------------------------------------------------------
+%% API
+%%------------------------------------------------------------------------------
+
+-spec add_handlers() -> ok.
+add_handlers() ->
+    %% HTTP API handler
+    ok = emqx_conf:add_handler([?CONF_KEY_ROOT, schemas, '?'], ?MODULE),
+    %% Conf load / data import handler
+    ok = emqx_conf:add_handler(?CONF_KEY_PATH, ?MODULE),
+    ok.
+
+-spec remove_handlers() -> ok.
+remove_handlers() ->
+    ok = emqx_conf:remove_handler([?CONF_KEY_ROOT, schemas, '?']),
+    ok = emqx_conf:remove_handler(?CONF_KEY_PATH),
+    ok.
+
+%%------------------------------------------------------------------------------
+%% `emqx_config_handler' API
+%%------------------------------------------------------------------------------
+
+%% remove
+post_config_update(
+    ?SCHEMA_CONF_PATH(Name),
+    '$remove',
+    _NewSchemas,
+    _OldSchemas,
+    _AppEnvs
+) ->
+    emqx_schema_registry:async_delete_serdes([Name]),
+    ok;
+%% add or update
+post_config_update(
+    ?SCHEMA_CONF_PATH(NewName),
+    _Cmd,
+    NewSchema,
+    OldSchema,
+    _AppEnvs
+) ->
+    case OldSchema of
+        undefined ->
+            ok;
+        _ ->
+            emqx_schema_registry:ensure_serde_absent(NewName)
+    end,
+    case emqx_schema_registry:build_serdes([{NewName, NewSchema}]) of
+        ok ->
+            {ok, #{NewName => NewSchema}};
+        {error, Reason, SerdesToRollback} ->
+            lists:foreach(fun emqx_schema_registry:ensure_serde_absent/1, SerdesToRollback),
+            {error, Reason}
+    end;
+post_config_update(?CONF_KEY_PATH, _Cmd, NewConf = #{schemas := NewSchemas}, OldConf, _AppEnvs) ->
+    OldSchemas = maps:get(schemas, OldConf, #{}),
+    #{
+        added := Added,
+        changed := Changed0,
+        removed := Removed
+    } = emqx_utils_maps:diff_maps(NewSchemas, OldSchemas),
+    Changed = maps:map(fun(_N, {_Old, New}) -> New end, Changed0),
+    RemovedNames = maps:keys(Removed),
+    case RemovedNames of
+        [] ->
+            ok;
+        _ ->
+            emqx_schema_registry:async_delete_serdes(RemovedNames)
+    end,
+    SchemasToBuild = maps:to_list(maps:merge(Changed, Added)),
+    ok = lists:foreach(fun emqx_schema_registry:ensure_serde_absent/1, [
+        N
+     || {N, _} <- SchemasToBuild
+    ]),
+    case emqx_schema_registry:build_serdes(SchemasToBuild) of
+        ok ->
+            {ok, NewConf};
+        {error, Reason, SerdesToRollback} ->
+            lists:foreach(fun emqx_schema_registry:ensure_serde_absent/1, SerdesToRollback),
+            {error, Reason}
+    end;
+post_config_update(_Path, _Cmd, NewConf, _OldConf, _AppEnvs) ->
+    {ok, NewConf}.
+
+%%------------------------------------------------------------------------------
+%% `emqx_config_backup' API
+%%------------------------------------------------------------------------------
+
+import_config(#{<<"schema_registry">> := #{<<"schemas">> := Schemas} = SchemaRegConf}) ->
+    OldSchemas = emqx:get_raw_config([?CONF_KEY_ROOT, schemas], #{}),
+    SchemaRegConf1 = SchemaRegConf#{<<"schemas">> => maps:merge(OldSchemas, Schemas)},
+    case emqx_conf:update(?CONF_KEY_PATH, SchemaRegConf1, #{override_to => cluster}) of
+        {ok, #{raw_config := #{<<"schemas">> := NewRawSchemas}}} ->
+            Changed = maps:get(changed, emqx_utils_maps:diff_maps(NewRawSchemas, OldSchemas)),
+            ChangedPaths = [[?CONF_KEY_ROOT, schemas, Name] || Name <- maps:keys(Changed)],
+            {ok, #{root_key => ?CONF_KEY_ROOT, changed => ChangedPaths}};
+        Error ->
+            {error, #{root_key => ?CONF_KEY_ROOT, reason => Error}}
+    end;
+import_config(_RawConf) ->
+    {ok, #{root_key => ?CONF_KEY_ROOT, changed => []}}.
+
+%%------------------------------------------------------------------------------
+%% Internal fns
+%%------------------------------------------------------------------------------

apps/emqx_schema_registry/test/emqx_schema_registry_SUITE.erl

@@ -787,11 +787,11 @@ t_import_config(_Config) ->
     Path = [schema_registry, schemas, <<"my_avro_schema">>],
     ?assertEqual(
         {ok, #{root_key => schema_registry, changed => []}},
-        emqx_schema_registry:import_config(RawConf)
+        emqx_schema_registry_config:import_config(RawConf)
     ),
     ?assertEqual(
         {ok, #{root_key => schema_registry, changed => [Path]}},
-        emqx_schema_registry:import_config(RawConf1)
+        emqx_schema_registry_config:import_config(RawConf1)
     ).
 
 sparkplug_example_data_base64() ->

apps/emqx_schema_registry/test/emqx_schema_registry_http_api_SUITE.erl

@@ -316,7 +316,7 @@ t_crud(Config) ->
         {ok, 400, #{
             <<"code">> := <<"BAD_REQUEST">>,
             <<"message">> :=
-                <<"{post_config_update,emqx_schema_registry,", _/binary>>
+                <<"{post_config_update,emqx_schema_registry_config,", _/binary>>
         }},
         request({put, SchemaName, UpdateParams#{<<"source">> := InvalidSourceBin}})
     ),
@@ -357,7 +357,7 @@ t_crud(Config) ->
         {ok, 400, #{
             <<"code">> := <<"BAD_REQUEST">>,
             <<"message">> :=
-                <<"{post_config_update,emqx_schema_registry,", _/binary>>
+                <<"{post_config_update,emqx_schema_registry_config,", _/binary>>
         }},
         request({post, Params#{<<"source">> := InvalidSourceBin}})
     ),

apps/emqx_schema_registry/test/emqx_schema_registry_serde_SUITE.erl

@@ -42,12 +42,15 @@ end_per_suite(Config) ->
     Apps = ?config(apps, Config),
     emqx_cth_suite:stop(Apps),
     ok.
+
 init_per_testcase(_TestCase, Config) ->
     Config.
 
 end_per_testcase(_TestCase, _Config) ->
     emqx_common_test_helpers:call_janitor(),
+    snabbkaffe:start_trace(),
     clear_schemas(),
+    snabbkaffe:stop(),
     ok.
 
 %%------------------------------------------------------------------------------