Add node column to list_stream_publishers/consumers

Fixes #9582

deps/rabbitmq_stream/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListStreamConsumersCommand.erl

@@ -11,7 +11,7 @@
 %% The Original Code is RabbitMQ.
 %%
 %% The Initial Developer of the Original Code is GoPivotal, Inc.
-%% Copyright (c) 2021 VMware, Inc. or its affiliates.  All rights reserved.
+%% Copyright (c) 2021-2023 VMware, Inc. or its affiliates.  All rights reserved.
 
 -module('Elixir.RabbitMQ.CLI.Ctl.Commands.ListStreamConsumersCommand').
 
@@ -65,7 +65,7 @@ validate(Args, _) ->
 
 merge_defaults([], Opts) ->
     merge_defaults([rabbit_data_coercion:to_binary(Item)
-                    || Item <- ?CONSUMER_INFO_ITEMS],
+                    || Item <- ?CONSUMER_INFO_ITEMS -- [connection_pid, node]],
                    Opts);
 merge_defaults(Args, Opts) ->
     {Args, maps:merge(#{verbose => false, vhost => <<"/">>}, Opts)}.

deps/rabbitmq_stream/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListStreamPublishersCommand.erl

@@ -11,7 +11,7 @@
 %% The Original Code is RabbitMQ.
 %%
 %% The Initial Developer of the Original Code is GoPivotal, Inc.
-%% Copyright (c) 2021 VMware, Inc. or its affiliates.  All rights reserved.
+%% Copyright (c) 2021-2023 VMware, Inc. or its affiliates.  All rights reserved.
 
 -module('Elixir.RabbitMQ.CLI.Ctl.Commands.ListStreamPublishersCommand').
 
@@ -65,7 +65,7 @@ validate(Args, _) ->
 
 merge_defaults([], Opts) ->
     merge_defaults([rabbit_data_coercion:to_binary(Item)
-                    || Item <- ?PUBLISHER_INFO_ITEMS],
+                    || Item <- ?PUBLISHER_INFO_ITEMS -- [connection_pid, node]],
                    Opts);
 merge_defaults(Args, Opts) ->
     {Args, maps:merge(#{verbose => false, vhost => <<"/">>}, Opts)}.

deps/rabbitmq_stream/src/rabbit_stream_reader.erl

@@ -3688,6 +3688,8 @@ consumer_i(offset_lag,
     stream_stored_offset(Log) - consumer_offset(Counters);
 consumer_i(connection_pid, _) ->
     self();
+consumer_i(node, _) ->
+    node();
 consumer_i(properties,
            #consumer{configuration =
                          #consumer_configuration{properties = Properties}}) ->
@@ -3720,6 +3722,8 @@ publisher_i(stream, #publisher{stream = S}) ->
     S;
 publisher_i(connection_pid, _) ->
     self();
+publisher_i(node, _) ->
+    node();
 publisher_i(publisher_id, #publisher{publisher_id = Id}) ->
     Id;
 publisher_i(reference, #publisher{reference = undefined}) ->

deps/rabbitmq_stream/test/commands_SUITE.erl

@@ -197,7 +197,7 @@ list_tls_connections_run(Config) ->
 list_consumers_merge_defaults(_Config) ->
     DefaultItems =
         [rabbit_data_coercion:to_binary(Item)
-         || Item <- ?CONSUMER_INFO_ITEMS],
+         || Item <- ?CONSUMER_INFO_ITEMS -- [connection_pid, node]],
     {DefaultItems, #{verbose := false}} =
         ?COMMAND_LIST_CONSUMERS:merge_defaults([], #{}),
 
@@ -266,7 +266,7 @@ list_consumers_run(Config) ->
 list_publishers_merge_defaults(_Config) ->
     DefaultItems =
         [rabbit_data_coercion:to_binary(Item)
-         || Item <- ?PUBLISHER_INFO_ITEMS],
+         || Item <- ?PUBLISHER_INFO_ITEMS -- [connection_pid, node]],
     {DefaultItems, #{verbose := false}} =
         ?COMMAND_LIST_PUBLISHERS:merge_defaults([], #{}),
 

deps/rabbitmq_stream_common/include/rabbit_stream.hrl

@@ -95,6 +95,7 @@
 
 -define(CONSUMER_INFO_ITEMS, [
   connection_pid,
+  node,
   subscription_id,
   stream,
   messages_consumed,
@@ -108,6 +109,7 @@
 
 -define(PUBLISHER_INFO_ITEMS, [
   connection_pid,
+  node,
   publisher_id,
   stream,
   reference,