Add unit for "simple" SAC group

"simple" meaning not part of a super stream.

References #3753

Author: acogoluegnes

deps/rabbit/src/rabbit_stream_sac_coordinator.erl

@@ -58,13 +58,7 @@ apply({register_consumer,
        ConnectionPid,
        SubscriptionId},
       #?MODULE{groups = StreamGroups0} = State) ->
-    %% TODO monitor connection PID to remove consumers when their connection dies
-    %% this could require some index to avoid crawling the whole data structure
-    %% this is necessary to fail over to another consumer when one dies abruptly
-    %% also, check the liveliness of each consumer whenever there's a change in the group,
-    %% to make sure to get rid of zombies
-    %%
-    %% TODO monitor streams and virtual hosts as well
+    %% FIXME monitor virtual hosts as well?
     rabbit_log:debug("New consumer ~p ~p in group ~p, partition index "
                      "is ~p",
                      [ConnectionPid,

deps/rabbit/test/rabbit_stream_sac_coordinator_SUITE.erl

@@ -57,6 +57,76 @@ init_per_testcase(_TestCase, Config) ->
 end_per_testcase(_TestCase, _Config) ->
     ok.
 
+simple_sac_test(_) ->
+    Stream = <<"stream">>,
+    ConsumerName = <<"app">>,
+    ConnectionPid = self(),
+    GroupId = {<<"/">>, Stream, ConsumerName},
+    Command0 =
+        register_consumer_command(Stream, -1, ConsumerName, ConnectionPid, 0),
+    State0 = state(),
+    {#?STATE{groups = #{GroupId := #group{consumers = Consumers1}}} =
+         State1,
+     {ok, Active1}, Effects1} =
+        rabbit_stream_sac_coordinator:apply(Command0, State0),
+    ?assert(Active1),
+    ?assertEqual([consumer(ConnectionPid, 0, true)], Consumers1),
+    assertSendMessageEffect(ConnectionPid, 0, true, Effects1),
+
+    Command1 =
+        register_consumer_command(Stream, -1, ConsumerName, ConnectionPid, 1),
+    {#?STATE{groups = #{GroupId := #group{consumers = Consumers2}}} =
+         State2,
+     {ok, Active2}, Effects2} =
+        rabbit_stream_sac_coordinator:apply(Command1, State1),
+    ?assertNot(Active2),
+    ?assertEqual([consumer(ConnectionPid, 0, true),
+                  consumer(ConnectionPid, 1, false)],
+                 Consumers2),
+    assertEmpty(Effects2),
+
+    Command2 =
+        register_consumer_command(Stream, -1, ConsumerName, ConnectionPid, 2),
+    {#?STATE{groups = #{GroupId := #group{consumers = Consumers3}}} =
+         State3,
+     {ok, Active3}, Effects3} =
+        rabbit_stream_sac_coordinator:apply(Command2, State2),
+    ?assertNot(Active3),
+    ?assertEqual([consumer(ConnectionPid, 0, true),
+                  consumer(ConnectionPid, 1, false),
+                  consumer(ConnectionPid, 2, false)],
+                 Consumers3),
+    assertEmpty(Effects3),
+
+    Command3 =
+        unregister_consumer_command(Stream, ConsumerName, ConnectionPid, 0),
+    {#?STATE{groups = #{GroupId := #group{consumers = Consumers4}}} =
+         State4,
+     ok, Effects4} =
+        rabbit_stream_sac_coordinator:apply(Command3, State3),
+    ?assertEqual([consumer(ConnectionPid, 1, true),
+                  consumer(ConnectionPid, 2, false)],
+                 Consumers4),
+    assertSendMessageEffect(ConnectionPid, 1, true, Effects4),
+
+    Command4 =
+        unregister_consumer_command(Stream, ConsumerName, ConnectionPid, 1),
+    {#?STATE{groups = #{GroupId := #group{consumers = Consumers5}}} =
+         State5,
+     ok, Effects5} =
+        rabbit_stream_sac_coordinator:apply(Command4, State4),
+    ?assertEqual([consumer(ConnectionPid, 2, true)], Consumers5),
+    assertSendMessageEffect(ConnectionPid, 2, true, Effects5),
+
+    Command5 =
+        unregister_consumer_command(Stream, ConsumerName, ConnectionPid, 2),
+    {#?STATE{groups = Groups6}, ok, Effects6} =
+        rabbit_stream_sac_coordinator:apply(Command5, State5),
+    assertEmpty(Groups6),
+    assertEmpty(Effects6),
+
+    ok.
+
 ensure_monitors_test(_) ->
     GroupId = {<<"/">>, <<"stream">>, <<"app">>},
     Group =
@@ -139,13 +209,7 @@ handle_connection_down_test(_) ->
         rabbit_stream_sac_coordinator:handle_connection_down(Pid0, State0),
     assertSize(1, PidsGroups1),
     assertSize(1, maps:get(Pid1, PidsGroups1)),
-    ?assertEqual([{mod_call,
-                   rabbit_stream_sac_coordinator,
-                   send_message,
-                   [Pid1,
-                    {sac,
-                     {{subscription_id, 1}, {active, true}, {extra, []}}}]}],
-                 Effects1),
+    assertSendMessageEffect(Pid1, 1, true, Effects1),
     ?assertEqual(#{GroupId => cgroup([consumer(Pid1, 1, true)])},
                  Groups1),
     {#?STATE{pids_groups = PidsGroups2, groups = Groups2} = _State2,
@@ -183,6 +247,9 @@ cgroup(Consumers) ->
 cgroup(PartitionIndex, Consumers) ->
     #group{partition_index = PartitionIndex, consumers = Consumers}.
 
+state() ->
+    state(#{}).
+
 state(Groups) ->
     state(Groups, #{}).
 
@@ -212,3 +279,13 @@ unregister_consumer_command(Stream,
      ConsumerName,
      ConnectionPid,
      SubId}.
+
+assertSendMessageEffect(Pid, SubId, Active, [Effect]) ->
+    ?assertEqual({mod_call,
+                  rabbit_stream_sac_coordinator,
+                  send_message,
+                  [Pid,
+                   {sac,
+                    {{subscription_id, SubId}, {active, Active},
+                     {extra, []}}}]},
+                 Effect).