feat(dssubs): make startup sequence dynamic + run only if enabled

apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_app.erl

@@ -15,11 +15,18 @@
 
 -spec start(application:start_type(), term()) -> {ok, pid()}.
 start(_Type, _Args) ->
-    ok = emqx_ds_shared_sub_config:load(),
     {ok, Sup} = emqx_ds_shared_sub_sup:start_link(),
+    ok = emqx_ds_shared_sub_config:load(),
+    case emqx_ds_shared_sub_config:enabled() of
+        true ->
+            ok = emqx_ds_shared_sub_sup:on_enable();
+        false ->
+            ok
+    end,
     {ok, Sup}.
 
 -spec stop(term()) -> ok.
 stop(_State) ->
+    ok = emqx_ds_shared_sub_sup:on_disable(),
     ok = emqx_ds_shared_sub_config:unload(),
     ok.

apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_config.erl

@@ -24,7 +24,8 @@
 -export([
     load/0,
     unload/0,
-    get/1
+    get/1,
+    enabled/0
 ]).
 
 %%--------------------------------------------------------------------
@@ -45,6 +46,10 @@ get(Name) when is_atom(Name) ->
 get(Name) when is_list(Name) ->
     emqx_config:get([durable_queues | Name]).
 
+-spec enabled() -> boolean().
+enabled() ->
+    emqx_persistent_message:is_persistence_enabled() andalso ?MODULE:get(enable).
+
 %%--------------------------------------------------------------------
 %% emqx_config_handler callbacks
 %%--------------------------------------------------------------------
@@ -62,8 +67,22 @@ pre_config_update([durable_queues | _], NewConfig, _OldConfig) ->
     emqx_config:app_envs()
 ) ->
     ok.
-post_config_update([durable_queues | _], _Req, _NewConfig, _OldConfig, _AppEnvs) ->
-    ok.
+post_config_update([durable_queues | _], _Req, NewConfig, OldConfig, _AppEnvs) ->
+    case config_transition(NewConfig, OldConfig) of
+        enable ->
+            emqx_ds_shared_sub_sup:on_enable();
+        disable ->
+            emqx_ds_shared_sub_sup:on_disable();
+        undefined ->
+            ok
+    end.
+
+config_transition(#{enable := true}, #{enable := false}) ->
+    enable;
+config_transition(#{enable := false}, #{enable := true}) ->
+    disable;
+config_transition(#{enable := E}, #{enable := E}) ->
+    undefined.
 
 %%----------------------------------------------------------------------------------------
 %% Data backup

apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_registry.erl

@@ -70,7 +70,6 @@ start_elector(ShareTopic, StartTime) ->
 %%------------------------------------------------------------------------------
 
 init([]) ->
-    ok = emqx_ds_shared_sub_leader_store:open(),
     SupFlags = #{
         strategy => one_for_one,
         intensity => 10,

apps/emqx_ds_shared_sub/src/emqx_ds_shared_sub_sup.erl

@@ -7,7 +7,11 @@
 -behaviour(supervisor).
 
 %% API
--export([start_link/0]).
+-export([
+    start_link/0,
+    on_enable/0,
+    on_disable/0
+]).
 
 %% supervisor behaviour callbacks
 -export([init/1]).
@@ -19,6 +23,36 @@
 start_link() ->
     supervisor:start_link({local, ?MODULE}, ?MODULE, []).
 
+on_enable() ->
+    ok = emqx_ds_shared_sub_leader_store:open(),
+    ensure_started(emqx_ds_shared_sub_registry:child_spec()).
+
+on_disable() ->
+    ok = ensure_stopped(emqx_ds_shared_sub_registry:child_spec()),
+    emqx_ds_shared_sub_leader_store:close().
+
+%%------------------------------------------------------------------------------
+
+ensure_started(ChildSpec) ->
+    case supervisor:start_child(?MODULE, ChildSpec) of
+        {ok, _Pid} ->
+            ok;
+        {error, {already_started, _}} ->
+            ok;
+        {error, _} = Error ->
+            Error
+    end.
+
+ensure_stopped(#{id := ChildId}) ->
+    case supervisor:terminate_child(?MODULE, ChildId) of
+        ok ->
+            supervisor:delete_child(?MODULE, ChildId);
+        {error, not_found} ->
+            ok;
+        {error, _} = Error ->
+            Error
+    end.
+
 %%------------------------------------------------------------------------------
 %% supervisor behaviour callbacks
 %%------------------------------------------------------------------------------
@@ -29,7 +63,4 @@ init([]) ->
         intensity => 10,
         period => 10
     },
-    ChildSpecs = [
-        emqx_ds_shared_sub_registry:child_spec()
-    ],
-    {ok, {SupFlags, ChildSpecs}}.
+    {ok, {SupFlags, []}}.