Add connectivity label to list_stream_group_consumers

Author: acogoluegnes

deps/rabbit/src/rabbit_stream_sac_coordinator.erl

@@ -152,6 +152,7 @@ wrap_cmd(Cmd) ->
     {sac, Cmd}.
 
 %% return the current groups for a given virtual host
+%% (CLI command)
 -spec consumer_groups(binary(), [atom()]) ->
                          {ok,
                           [term()]} | {error, sac_error() | term()}.
@@ -173,6 +174,7 @@ consumer_groups(VirtualHost, InfoKeys) ->
     end.
 
 %% get the consumers of a given group in a given virtual host
+%% (CLI command)
 -spec group_consumers(binary(), binary(), binary(), [atom()]) ->
                          {ok, [term()]} |
                          {error, sac_error() | term()}.
@@ -544,6 +546,7 @@ maybe_rebalance_group(#group{partition_index = _, consumers = Consumers} = G,
             end
     end.
 
+%% used by CLI
 -spec consumer_groups(binary(), [atom()], state()) -> {ok, [term()]}.
 consumer_groups(VirtualHost, InfoKeys, #?MODULE{groups = Groups} = S)
   when ?IS_STATE_REC(S) ->
@@ -586,38 +589,14 @@ consumer_groups(VirtualHost, InfoKeys, S) ->
                       [atom()],
                       state()) ->
                          {ok, [term()]} | {error, not_found}.
-group_consumers(VirtualHost,
-                Stream,
-                Reference,
-                InfoKeys,
+group_consumers(VH, St, Ref, InfoKeys,
                 #?MODULE{groups = Groups} = S)
   when ?IS_STATE_REC(S) ->
-    GroupId = {VirtualHost, Stream, Reference},
+    GroupId = {VH, St, Ref},
     case Groups of
         #{GroupId := #group{consumers = Consumers}} ->
-            Cs = lists:foldr(fun(#consumer{subscription_id = SubId,
-                                           owner = Owner,
-                                           status = Status},
-                                 Acc) ->
-                                Record =
-                                    lists:foldr(fun (subscription_id, RecAcc) ->
-                                                        [{subscription_id,
-                                                          SubId}
-                                                         | RecAcc];
-                                                    (connection_name, RecAcc) ->
-                                                        [{connection_name,
-                                                          Owner}
-                                                         | RecAcc];
-                                                    (state, RecAcc) ->
-                                                        [{state, cli_consumer_status_label(Status)}
-                                                         | RecAcc];
-                                                    (Unknown, RecAcc) ->
-                                                        [{Unknown,
-                                                          unknown_field}
-                                                         | RecAcc]
-                                                end,
-                                                [], InfoKeys),
-                                [Record | Acc]
+            Cs = lists:foldr(fun(C, Acc) ->
+                                     [csr_cli_record(C, InfoKeys) | Acc]
                              end,
                              [], Consumers),
             {ok, Cs};
@@ -628,12 +607,27 @@ group_consumers(VirtualHost, Stream, Reference, InfoKeys, S) ->
     rabbit_stream_sac_coordinator_v4:group_consumers(VirtualHost, Stream,
                                                      Reference, InfoKeys, S).
 
-cli_consumer_status_label({?PDOWN, _}) ->
-   inactive;
-cli_consumer_status_label({_, ?ACTIVE}) ->
-   active;
-cli_consumer_status_label(_) ->
-   inactive.
+csr_cli_record(#consumer{subscription_id = SubId, owner = Owner,
+                         status = Status}, InfoKeys) ->
+    lists:foldr(fun (subscription_id, Acc) ->
+                        [{subscription_id, SubId} | Acc];
+                    (connection_name, Acc) ->
+                        [{connection_name, Owner} | Acc];
+                    (state, Acc) ->
+                        [{state, cli_csr_status_label(Status)} | Acc];
+                    (Unknown, Acc) ->
+                        [{Unknown, unknown_field} | Acc]
+                end,
+                [], InfoKeys).
+
+
+cli_csr_status_label({Cnty, Acty}) ->
+   rabbit_data_coercion:to_list(Acty) ++ " (" ++ connectivity_label(Cnty) ++ ")".
+
+connectivity_label(?PDOWN) ->
+    "presumed down";
+connectivity_label(Cnty) ->
+    rabbit_data_coercion:to_list(Cnty).
 
 -spec ensure_monitors(command(),
                       state(),

deps/rabbitmq_stream/test/commands_SUITE.erl

@@ -378,7 +378,7 @@ list_consumer_groups_run(Config) ->
     {ok, []} = ?COMMAND_LIST_CONSUMER_GROUPS:run([], Opts),
 
     StreamPort = rabbit_stream_SUITE:get_stream_port(Config),
-    {S, C} = start_stream_connection(StreamPort),
+    {S, C0} = start_stream_connection(StreamPort),
     ?awaitMatch(1, connection_count(Config), ?WAIT),
 
     ConsumerReference = <<"foo">>,
@@ -387,34 +387,34 @@ list_consumer_groups_run(Config) ->
           <<"name">> => ConsumerReference},
 
     Stream1 = <<"list_consumer_groups_run_1">>,
-    create_stream(S, Stream1, C),
-    subscribe(S, 0, Stream1, SubProperties, C),
-    handle_consumer_update(S, C, 0),
-    subscribe(S, 1, Stream1, SubProperties, C),
-    subscribe(S, 2, Stream1, SubProperties, C),
+    C1 = create_stream(S, Stream1, C0),
+    C2 = subscribe(S, 0, Stream1, SubProperties, C1),
+    C3 = handle_consumer_update(S, C2, 0),
+    C4 = subscribe(S, 1, Stream1, SubProperties, C3),
+    C5 = subscribe(S, 2, Stream1, SubProperties, C4),
 
     ?awaitMatch(3, consumer_count(Config), ?WAIT),
 
     {ok, [CG1]} = ?COMMAND_LIST_CONSUMER_GROUPS:run([], Opts),
     assertConsumerGroup(Stream1, ConsumerReference, -1, 3, CG1),
 
     Stream2 = <<"list_consumer_groups_run_2">>,
-    create_stream(S, Stream2, C),
-    subscribe(S, 3, Stream2, SubProperties, C),
-    handle_consumer_update(S, C, 3),
-    subscribe(S, 4, Stream2, SubProperties, C),
-    subscribe(S, 5, Stream2, SubProperties, C),
+    C6 = create_stream(S, Stream2, C5),
+    C7 = subscribe(S, 3, Stream2, SubProperties, C6),
+    C8 = handle_consumer_update(S, C7, 3),
+    C9 = subscribe(S, 4, Stream2, SubProperties, C8),
+    C10 = subscribe(S, 5, Stream2, SubProperties, C9),
 
     ?awaitMatch(3 + 3, consumer_count(Config), ?WAIT),
 
     {ok, [CG1, CG2]} = ?COMMAND_LIST_CONSUMER_GROUPS:run([], Opts),
     assertConsumerGroup(Stream1, ConsumerReference, -1, 3, CG1),
     assertConsumerGroup(Stream2, ConsumerReference, -1, 3, CG2),
 
-    delete_stream(S, Stream1, C),
-    delete_stream(S, Stream2, C),
+    C11 = delete_stream(S, Stream1, C10),
+    C12 = delete_stream(S, Stream2, C11),
 
-    close(S, C),
+    close(S, C12),
     {ok, []} = ?COMMAND_LIST_CONSUMER_GROUPS:run([], Opts),
     ok.
 
@@ -490,9 +490,9 @@ list_group_consumers_run(Config) ->
 
     {ok, Consumers1} =
         ?COMMAND_LIST_GROUP_CONSUMERS:run(Args, OptsGroup1),
-    ?assertEqual([[{subscription_id, 0}, {state, active}],
-                  [{subscription_id, 1}, {state, inactive}],
-                  [{subscription_id, 2}, {state, inactive}]],
+    ?assertEqual([[{subscription_id, 0}, {state, "active (connected)"}],
+                  [{subscription_id, 1}, {state, "waiting (connected)"}],
+                  [{subscription_id, 2}, {state, "waiting (connected)"}]],
                  Consumers1),
 
     Stream2 = <<"list_group_consumers_run_2">>,
@@ -510,9 +510,9 @@ list_group_consumers_run(Config) ->
 
     {ok, Consumers2} =
         ?COMMAND_LIST_GROUP_CONSUMERS:run(Args, OptsGroup2),
-    ?assertEqual([[{subscription_id, 3}, {state, active}],
-                  [{subscription_id, 4}, {state, inactive}],
-                  [{subscription_id, 5}, {state, inactive}]],
+    ?assertEqual([[{subscription_id, 3}, {state, "active (connected)"}],
+                  [{subscription_id, 4}, {state, "waiting (connected)"}],
+                  [{subscription_id, 5}, {state, "waiting (connected)"}]],
                  Consumers2),
 
     delete_stream(S, Stream1, C),