Move SAC API to SAC coordinator

As per @ansd's suggestion. This way the stream coordinator
does not have to know about the SAC commands.

References #3753

5 files changed, 160 insertions, 119 deletions

diff --git a/deps/rabbit/src/rabbit_stream_coordinator.erl b/deps/rabbit/src/rabbit_stream_coordinator.erl
index 9ea98e8c00..a36d4cc928 100644
--- a/deps/rabbit/src/rabbit_stream_coordinator.erl
+++ b/deps/rabbit/src/rabbit_stream_coordinator.erl
@@ -40,17 +40,14 @@
 -export([log_overview/1]).
 -export([replay/1]).
 
+%% for SAC coordinator
+-export([process_command/1,
+         sac_state/1]).
+
 %% for testing and debugging
 -export([eval_listeners/3,
          state/0]).
 
-%% Single Active Consumer API
--export([register_consumer/7,
-         unregister_consumer/5,
-         activate_consumer/3,
-         consumer_groups/2,
-         group_consumers/4]).
-
 -rabbit_boot_step({?MODULE,
                    [{description, "Restart stream coordinator"},
                     {mfa,         {?MODULE, recover, []}},
@@ -179,6 +176,9 @@ policy_changed(Q) when ?is_amqqueue(Q) ->
     StreamId = maps:get(name, amqqueue:get_type_state(Q)),
     process_command({policy_changed, StreamId, #{queue => Q}}).
 
+sac_state(#?MODULE{single_active_consumer = SacState}) ->
+    SacState.
+
 %% for debugging
 state() ->
     case ra:local_query({?MODULE, node()}, fun(State) -> State end) of
@@ -281,101 +281,6 @@ register_local_member_listener(Q) when ?is_amqqueue(Q) ->
                        stream_id => StreamId,
                        type => local_member}}).
 
-%% Single Active Consumer API
--spec register_consumer(binary(), binary(), integer(), binary(), pid(), binary(), integer()) ->
-    {ok, boolean()} | {error, feature_flag_disabled}.
-register_consumer(VirtualHost,
-                  Stream,
-                  PartitionIndex,
-                  ConsumerName,
-                  ConnectionPid,
-                  Owner,
-                  SubscriptionId) ->
-    maybe_sac_execute(
-      fun() ->
-              {ok, Res, _} = process_command({sac,
-                                              {register_consumer, VirtualHost, Stream, PartitionIndex, ConsumerName,
-                                               ConnectionPid, Owner, SubscriptionId}}),
-              Res
-      end).
-
--spec unregister_consumer(binary(), binary(), binary(), pid(), integer()) -> ok | {error, feature_flag_disabled}.
-unregister_consumer(VirtualHost,
-                    Stream,
-                    ConsumerName,
-                    ConnectionPid,
-                    SubscriptionId) ->
-    maybe_sac_execute(
-      fun() ->
-              {ok, Res, _} = process_command({sac,
-                                              {unregister_consumer, VirtualHost, Stream, ConsumerName, ConnectionPid, SubscriptionId}}),
-              Res
-      end).
-
--spec activate_consumer(binary(), binary(), binary()) -> ok | {error, feature_flag_disabled}.
-activate_consumer(VirtualHost, Stream, ConsumerName) ->
-    maybe_sac_execute(
-      fun() ->
-              {ok, Res, _} = process_command({sac, {activate_consumer, VirtualHost, Stream, ConsumerName}}),
-              Res
-      end).
-
-%% return the current groups for a given virtual host
--spec consumer_groups(binary(), [atom()]) -> {ok, [term()] | {error, feature_flag_disabled | atom()}}.
-consumer_groups(VirtualHost, InfoKeys) ->
-    maybe_sac_execute(
-      fun() ->
-              case ra:local_query({?MODULE, node()},
-                                  fun(#?MODULE{single_active_consumer = SacState}) ->
-                                          rabbit_stream_sac_coordinator:consumer_groups(
-                                            VirtualHost, InfoKeys, SacState)
-                                  end) of
-                  {ok, {_, Result}, _} ->
-                      Result;
-                  {error, noproc} ->
-                      %% not started yet, so no groups
-                      {ok, []};
-                  {error, _} = Err ->
-                      Err;
-                  {timeout, _} ->
-                      {error, timeout}
-              end
-      end).
-
-%% get the consumers of a given group in a given virtual host
--spec group_consumers(binary(), binary(), binary(), [atom()]) ->
-    {ok, [term()]} | {error, feature_flag_disabled | atom()}.
-group_consumers(VirtualHost, Stream, Reference, InfoKeys) ->
-    maybe_sac_execute(
-      fun() ->
-              case ra:local_query({?MODULE, node()},
-                                  fun(#?MODULE{single_active_consumer = SacState}) ->
-                                          rabbit_stream_sac_coordinator:group_consumers(
-                                            VirtualHost, Stream, Reference,
-                                            InfoKeys, SacState)
-                                  end) of
-                  {ok, {_, {ok, _} = Result}, _} ->
-                      Result;
-                  {ok, {_, {error, _} = Err}, _} ->
-                      Err;
-                  {error, noproc} ->
-                      %% not started yet, so the group cannot exist
-                      {error, not_found};
-                  {error, _} = Err ->
-                      Err;
-                  {timeout, _} ->
-                      {error, timeout}
-              end
-      end).
-
-maybe_sac_execute(Fun) ->
-    case rabbit_stream_sac_coordinator:is_ff_enabled() of
-        true ->
-            Fun();
-        false ->
-            {error, feature_flag_disabled}
-    end.
-
 process_command(Cmd) ->
     Servers = ensure_coordinator_started(),
     process_command(Servers, Cmd).
diff --git a/deps/rabbit/src/rabbit_stream_sac_coordinator.erl b/deps/rabbit/src/rabbit_stream_sac_coordinator.erl
index 5f1e566bf9..9570549478 100644
--- a/deps/rabbit/src/rabbit_stream_sac_coordinator.erl
+++ b/deps/rabbit/src/rabbit_stream_sac_coordinator.erl
@@ -39,6 +39,12 @@
 -export_type([state/0,
               command/0]).
 
+%% Single Active Consumer API
+-export([register_consumer/7,
+         unregister_consumer/5,
+         activate_consumer/3,
+         consumer_groups/2,
+         group_consumers/4]).
 -export([apply/2,
          init_state/0,
          send_message/2,
@@ -48,6 +54,136 @@
          group_consumers/5,
          is_ff_enabled/0]).
 
+%% Single Active Consumer API
+-spec register_consumer(binary(),
+                        binary(),
+                        integer(),
+                        binary(),
+                        pid(),
+                        binary(),
+                        integer()) ->
+                           {ok, boolean()} | {error, feature_flag_disabled}.
+register_consumer(VirtualHost,
+                  Stream,
+                  PartitionIndex,
+                  ConsumerName,
+                  ConnectionPid,
+                  Owner,
+                  SubscriptionId) ->
+    maybe_sac_execute(fun() ->
+                         {ok, Res, _} =
+                             process_command({sac,
+                                              {register_consumer,
+                                               VirtualHost,
+                                               Stream,
+                                               PartitionIndex,
+                                               ConsumerName,
+                                               ConnectionPid,
+                                               Owner,
+                                               SubscriptionId}}),
+                         Res
+                      end).
+
+-spec unregister_consumer(binary(),
+                          binary(),
+                          binary(),
+                          pid(),
+                          integer()) ->
+                             ok | {error, feature_flag_disabled}.
+unregister_consumer(VirtualHost,
+                    Stream,
+                    ConsumerName,
+                    ConnectionPid,
+                    SubscriptionId) ->
+    maybe_sac_execute(fun() ->
+                         {ok, Res, _} =
+                             process_command({sac,
+                                              {unregister_consumer,
+                                               VirtualHost,
+                                               Stream,
+                                               ConsumerName,
+                                               ConnectionPid,
+                                               SubscriptionId}}),
+                         Res
+                      end).
+
+-spec activate_consumer(binary(), binary(), binary()) ->
+                           ok | {error, feature_flag_disabled}.
+activate_consumer(VirtualHost, Stream, ConsumerName) ->
+    maybe_sac_execute(fun() ->
+                         {ok, Res, _} =
+                             process_command({sac,
+                                              {activate_consumer,
+                                               VirtualHost,
+                                               Stream,
+                                               ConsumerName}}),
+                         Res
+                      end).
+
+process_command(Cmd) ->
+    rabbit_stream_coordinator:process_command(Cmd).
+
+%% return the current groups for a given virtual host
+-spec consumer_groups(binary(), [atom()]) ->
+                         {ok,
+                          [term()] | {error, feature_flag_disabled | atom()}}.
+consumer_groups(VirtualHost, InfoKeys) ->
+    maybe_sac_execute(fun() ->
+                         case ra:local_query({rabbit_stream_coordinator,
+                                              node()},
+                                             fun(State) ->
+                                                SacState =
+                                                    rabbit_stream_coordinator:sac_state(State),
+                                                consumer_groups(VirtualHost,
+                                                                InfoKeys,
+                                                                SacState)
+                                             end)
+                         of
+                             {ok, {_, Result}, _} -> Result;
+                             {error, noproc} ->
+                                 %% not started yet, so no groups
+                                 {ok, []};
+                             {error, _} = Err -> Err;
+                             {timeout, _} -> {error, timeout}
+                         end
+                      end).
+
+%% get the consumers of a given group in a given virtual host
+-spec group_consumers(binary(), binary(), binary(), [atom()]) ->
+                         {ok, [term()]} |
+                         {error, feature_flag_disabled | atom()}.
+group_consumers(VirtualHost, Stream, Reference, InfoKeys) ->
+    maybe_sac_execute(fun() ->
+                         case ra:local_query({rabbit_stream_coordinator,
+                                              node()},
+                                             fun(State) ->
+                                                SacState =
+                                                    rabbit_stream_coordinator:sac_state(State),
+                                                group_consumers(VirtualHost,
+                                                                Stream,
+                                                                Reference,
+                                                                InfoKeys,
+                                                                SacState)
+                                             end)
+                         of
+                             {ok, {_, {ok, _} = Result}, _} -> Result;
+                             {ok, {_, {error, _} = Err}, _} -> Err;
+                             {error, noproc} ->
+                                 %% not started yet, so the group cannot exist
+                                 {error, not_found};
+                             {error, _} = Err -> Err;
+                             {timeout, _} -> {error, timeout}
+                         end
+                      end).
+
+maybe_sac_execute(Fun) ->
+    case rabbit_stream_sac_coordinator:is_ff_enabled() of
+        true ->
+            Fun();
+        false ->
+            {error, feature_flag_disabled}
+    end.
+
 -spec init_state() -> state().
 init_state() ->
     #?MODULE{groups = #{}, pids_groups = #{}}.
diff --git a/deps/rabbitmq_stream/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListStreamConsumerGroupsCommand.erl b/deps/rabbitmq_stream/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListStreamConsumerGroupsCommand.erl
index 01e224d006..5a0f8b5dcb 100644
--- a/deps/rabbitmq_stream/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListStreamConsumerGroupsCommand.erl
+++ b/deps/rabbitmq_stream/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListStreamConsumerGroupsCommand.erl
@@ -103,7 +103,7 @@ run(Args,
         end,
 
     rabbit_misc:rpc_call(NodeName,
-                         rabbit_stream_coordinator,
+                         rabbit_stream_sac_coordinator,
                          consumer_groups,
                          [VHost, InfoKeys],
                          Timeout).
diff --git a/deps/rabbitmq_stream/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListStreamGroupConsumersCommand.erl b/deps/rabbitmq_stream/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListStreamGroupConsumersCommand.erl
index 52f555d730..390b1fc7e2 100644
--- a/deps/rabbitmq_stream/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListStreamGroupConsumersCommand.erl
+++ b/deps/rabbitmq_stream/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListStreamGroupConsumersCommand.erl
@@ -108,7 +108,7 @@ run(Args,
         end,
 
     rabbit_misc:rpc_call(NodeName,
-                         rabbit_stream_coordinator,
+                         rabbit_stream_sac_coordinator,
                          group_consumers,
                          [VHost, Stream, Reference, InfoKeys],
                          Timeout).
diff --git a/deps/rabbitmq_stream/src/rabbit_stream_reader.erl b/deps/rabbitmq_stream/src/rabbit_stream_reader.erl
index b6c40a3daf..a3475d95e4 100644
--- a/deps/rabbitmq_stream/src/rabbit_stream_reader.erl
+++ b/deps/rabbitmq_stream/src/rabbit_stream_reader.erl
@@ -2590,9 +2590,9 @@ handle_frame_post_auth(Transport,
                         case Extra of
                             [{stepping_down, true}] ->
                                 ConsumerName = consumer_name(Properties),
-                                rabbit_stream_coordinator:activate_consumer(VirtualHost,
-                                                                            Stream,
-                                                                            ConsumerName);
+                                rabbit_stream_sac_coordinator:activate_consumer(VirtualHost,
+                                                                                Stream,
+                                                                                ConsumerName);
                             _ ->
                                 ok
                         end,
@@ -2765,13 +2765,13 @@ maybe_register_consumer(VirtualHost,
                         true) ->
     PartitionIndex = partition_index(VirtualHost, Stream, Properties),
     {ok, Active} =
-        rabbit_stream_coordinator:register_consumer(VirtualHost,
-                                                    Stream,
-                                                    PartitionIndex,
-                                                    ConsumerName,
-                                                    self(),
-                                                    ConnectionName,
-                                                    SubscriptionId),
+        rabbit_stream_sac_coordinator:register_consumer(VirtualHost,
+                                                        Stream,
+                                                        PartitionIndex,
+                                                        ConsumerName,
+                                                        self(),
+                                                        ConnectionName,
+                                                        SubscriptionId),
     Active.
 
 maybe_notify_consumer(_, Connection, _, _, false = _Sac, _) ->
@@ -2812,11 +2812,11 @@ maybe_unregister_consumer(VirtualHost,
                                                                     SubscriptionId}},
                           true = _Sac) ->
     ConsumerName = consumer_name(Properties),
-    rabbit_stream_coordinator:unregister_consumer(VirtualHost,
-                                                  Stream,
-                                                  ConsumerName,
-                                                  self(),
-                                                  SubscriptionId).
+    rabbit_stream_sac_coordinator:unregister_consumer(VirtualHost,
+                                                      Stream,
+                                                      ConsumerName,
+                                                      self(),
+                                                      SubscriptionId).
 
 partition_index(VirtualHost, Stream, Properties) ->
     case Properties of