Fix keys for route and partitions responses

These 2 responses were not using the `rabbit_stream_core` utilities
so they were not using the "prefix" for responses.

Fixes #5956

5 files changed, 55 insertions, 88 deletions

diff --git a/deps/rabbitmq_stream/src/rabbit_stream.erl b/deps/rabbitmq_stream/src/rabbit_stream.erl
index 4a03b95379..47b487539c 100644
--- a/deps/rabbitmq_stream/src/rabbit_stream.erl
+++ b/deps/rabbitmq_stream/src/rabbit_stream.erl
@@ -207,8 +207,7 @@ emit_publisher_info_local(VHost, Items, Ref, AggregatorPid) ->
 
 list(VHost) ->
     [Client
-     || {_, ListSup, _, _}
-            <- supervisor:which_children(rabbit_stream_sup),
+     || {_, ListSup, _, _} <- supervisor:which_children(rabbit_stream_sup),
         {_, RanchEmbeddedSup, supervisor, _}
             <- supervisor:which_children(ListSup),
         {{ranch_listener_sup, _}, RanchListSup, _, _}
diff --git a/deps/rabbitmq_stream/src/rabbit_stream_connection_sup.erl b/deps/rabbitmq_stream/src/rabbit_stream_connection_sup.erl
index f583151d56..ccd2dc8846 100644
--- a/deps/rabbitmq_stream/src/rabbit_stream_connection_sup.erl
+++ b/deps/rabbitmq_stream/src/rabbit_stream_connection_sup.erl
@@ -21,40 +21,34 @@
 
 -include_lib("rabbit_common/include/rabbit.hrl").
 
--export([
-    start_link/3,
-    start_keepalive_link/0
-]).
+-export([start_link/3,
+         start_keepalive_link/0]).
 -export([init/1]).
 
 start_link(Ref, Transport, Opts) ->
     {ok, SupPid} = supervisor:start_link(?MODULE, []),
     {ok, KeepaliveSup} =
-        supervisor:start_child(
-            SupPid,
-            #{
-                id => rabbit_stream_keepalive_sup,
-                start => {rabbit_stream_connection_sup, start_keepalive_link, []},
-                restart => transient,
-                significant => true,
-                shutdown => infinity,
-                type => supervisor,
-                modules => [rabbit_keepalive_sup]
-            }
-        ),
+        supervisor:start_child(SupPid,
+                               #{id => rabbit_stream_keepalive_sup,
+                                 start =>
+                                     {rabbit_stream_connection_sup,
+                                      start_keepalive_link, []},
+                                 restart => transient,
+                                 significant => true,
+                                 shutdown => infinity,
+                                 type => supervisor,
+                                 modules => [rabbit_keepalive_sup]}),
     {ok, ReaderPid} =
-        supervisor:start_child(
-            SupPid,
-            #{
-                id => rabbit_stream_reader,
-                start => {rabbit_stream_reader, start_link, [KeepaliveSup, Transport, Ref, Opts]},
-                restart => transient,
-                significant => true,
-                shutdown => ?WORKER_WAIT,
-                type => worker,
-                modules => [rabbit_stream_reader]
-            }
-        ),
+        supervisor:start_child(SupPid,
+                               #{id => rabbit_stream_reader,
+                                 start =>
+                                     {rabbit_stream_reader, start_link,
+                                      [KeepaliveSup, Transport, Ref, Opts]},
+                                 restart => transient,
+                                 significant => true,
+                                 shutdown => ?WORKER_WAIT,
+                                 type => worker,
+                                 modules => [rabbit_stream_reader]}),
     {ok, SupPid, ReaderPid}.
 
 start_keepalive_link() ->
@@ -64,12 +58,8 @@ start_keepalive_link() ->
 
 init([]) ->
     {ok,
-        {
-            #{
-                strategy => one_for_all,
-                intensity => 0,
-                period => 1,
-                auto_shutdown => any_significant
-            },
-            []
-        }}.
+     {#{strategy => one_for_all,
+        intensity => 0,
+        period => 1,
+        auto_shutdown => any_significant},
+      []}}.
diff --git a/deps/rabbitmq_stream/src/rabbit_stream_reader.erl b/deps/rabbitmq_stream/src/rabbit_stream_reader.erl
index 533cc0b2f7..5a57f1b153 100644
--- a/deps/rabbitmq_stream/src/rabbit_stream_reader.erl
+++ b/deps/rabbitmq_stream/src/rabbit_stream_reader.erl
@@ -1880,7 +1880,6 @@ handle_frame_post_auth(Transport,
                                               Properties]),
                             Sac = single_active_consumer(Properties),
                             ConsumerName = consumer_name(Properties),
-                            %% TODO check consumer name is defined when SAC
                             case {Sac, rabbit_stream_utils:is_sac_ff_enabled(),
                                   ConsumerName}
                             of
@@ -2425,35 +2424,25 @@ handle_frame_post_auth(Transport,
                        State,
                        {request, CorrelationId,
                         {route, RoutingKey, SuperStream}}) ->
-    {ResponseCode, StreamBin} =
+    {ResponseCode, Streams} =
         case rabbit_stream_manager:route(RoutingKey, VirtualHost, SuperStream)
         of
             {ok, no_route} ->
-                {?RESPONSE_CODE_OK, <<0:32>>};
-            {ok, Streams} ->
-                StreamCount = length(Streams),
-                Bin = lists:foldl(fun(Stream, Acc) ->
-                                     StreamSize = byte_size(Stream),
-                                     <<Acc/binary, StreamSize:16,
-                                       Stream:StreamSize/binary>>
-                                  end,
-                                  <<StreamCount:32>>, Streams),
-                {?RESPONSE_CODE_OK, Bin};
+                {?RESPONSE_CODE_OK, []};
+            {ok, Strs} ->
+                {?RESPONSE_CODE_OK, Strs};
             {error, _} ->
                 rabbit_global_counters:increase_protocol_counter(stream,
                                                                  ?STREAM_DOES_NOT_EXIST,
                                                                  1),
-                {?RESPONSE_CODE_STREAM_DOES_NOT_EXIST, <<0:32>>}
+                {?RESPONSE_CODE_STREAM_DOES_NOT_EXIST, []}
         end,
 
     Frame =
-        <<?COMMAND_ROUTE:16,
-          ?VERSION_1:16,
-          CorrelationId:32,
-          ResponseCode:16,
-          StreamBin/binary>>,
-    FrameSize = byte_size(Frame),
-    Transport:send(S, <<FrameSize:32, Frame/binary>>),
+        rabbit_stream_core:frame({response, CorrelationId,
+                                  {route, ResponseCode, Streams}}),
+
+    Transport:send(S, Frame),
     {Connection, State};
 handle_frame_post_auth(Transport,
                        #stream_connection{socket = S,
@@ -2461,34 +2450,22 @@ handle_frame_post_auth(Transport,
                            Connection,
                        State,
                        {request, CorrelationId, {partitions, SuperStream}}) ->
-    {ResponseCode, PartitionsBin} =
+    {ResponseCode, Partitions} =
         case rabbit_stream_manager:partitions(VirtualHost, SuperStream) of
-            {ok, []} ->
-                {?RESPONSE_CODE_OK, <<0:32>>};
             {ok, Streams} ->
-                StreamCount = length(Streams),
-                Bin = lists:foldl(fun(Stream, Acc) ->
-                                     StreamSize = byte_size(Stream),
-                                     <<Acc/binary, StreamSize:16,
-                                       Stream:StreamSize/binary>>
-                                  end,
-                                  <<StreamCount:32>>, Streams),
-                {?RESPONSE_CODE_OK, Bin};
+                {?RESPONSE_CODE_OK, Streams};
             {error, _} ->
                 rabbit_global_counters:increase_protocol_counter(stream,
                                                                  ?STREAM_DOES_NOT_EXIST,
                                                                  1),
-                {?RESPONSE_CODE_STREAM_DOES_NOT_EXIST, <<0:32>>}
+                {?RESPONSE_CODE_STREAM_DOES_NOT_EXIST, []}
         end,
 
     Frame =
-        <<?COMMAND_PARTITIONS:16,
-          ?VERSION_1:16,
-          CorrelationId:32,
-          ResponseCode:16,
-          PartitionsBin/binary>>,
-    FrameSize = byte_size(Frame),
-    Transport:send(S, <<FrameSize:32, Frame/binary>>),
+        rabbit_stream_core:frame({response, CorrelationId,
+                                  {partitions, ResponseCode, Partitions}}),
+
+    Transport:send(S, Frame),
     {Connection, State};
 handle_frame_post_auth(Transport,
                        #stream_connection{transport = ConnTransport,
@@ -3367,10 +3344,8 @@ send_chunks(DeliverVersion,
 
 emit_stats(#stream_connection{publishers = Publishers} = Connection,
            #stream_connection_state{consumers = Consumers} = ConnectionState) ->
-    [{_, Pid},
-     {_, Recv_oct},
-     {_, Send_oct},
-     {_, Reductions}] = infos(?SIMPLE_METRICS, Connection, ConnectionState),
+    [{_, Pid}, {_, Recv_oct}, {_, Send_oct}, {_, Reductions}] =
+        infos(?SIMPLE_METRICS, Connection, ConnectionState),
     Infos = infos(?OTHER_METRICS, Connection, ConnectionState),
     rabbit_core_metrics:connection_stats(Pid, Infos),
     rabbit_core_metrics:connection_stats(Pid,
diff --git a/deps/rabbitmq_stream_common/src/rabbit_stream_core.erl b/deps/rabbitmq_stream_common/src/rabbit_stream_core.erl
index da0c0c3fba..41cee6923a 100644
--- a/deps/rabbitmq_stream_common/src/rabbit_stream_core.erl
+++ b/deps/rabbitmq_stream_common/src/rabbit_stream_core.erl
@@ -134,7 +134,7 @@
      {tune, FrameMax :: non_neg_integer(),
       HeartBeat :: non_neg_integer()} |
      {credit, response_code(), subscription_id()} |
-     {route, response_code(), stream_name()} |
+     {route, response_code(), [stream_name()]} |
      {partitions, response_code(), [stream_name()]} |
      {consumer_update, response_code(), none | offset_spec()} |
      {exchange_command_versions, response_code(),
@@ -442,8 +442,9 @@ response_body({metadata = Tag, Endpoints, Metadata}) ->
 
     NumStreams = map_size(Metadata),
     {command_id(Tag), [EndpointsBin, <<NumStreams:32>>, MetadataBin]};
-response_body({route = Tag, Code, Stream}) ->
-    {command_id(Tag), <<Code:16, ?STRING(Stream)>>};
+response_body({route = Tag, Code, Streams}) ->
+    StreamsBin = [<<?STRING(Stream)>> || Stream <- Streams],
+    {command_id(Tag), [<<Code:16, (length(Streams)):32>>, StreamsBin]};
 response_body({partitions = Tag, Code, Streams}) ->
     StreamsBin = [<<?STRING(Stream)>> || Stream <- Streams],
     {command_id(Tag), [<<Code:16, (length(Streams)):32>>, StreamsBin]};
@@ -934,8 +935,9 @@ parse_response_body(?COMMAND_SASL_AUTHENTICATE,
         end,
     {sasl_authenticate, ResponseCode, Challenge};
 parse_response_body(?COMMAND_ROUTE,
-                    <<ResponseCode:16, ?STRING(StreamSize, Stream)>>) ->
-    {route, ResponseCode, Stream};
+                    <<ResponseCode:16, _Count:32, StreamsBin/binary>>) ->
+    Streams = list_of_strings(StreamsBin),
+    {route, ResponseCode, Streams};
 parse_response_body(?COMMAND_PARTITIONS,
                     <<ResponseCode:16, _Count:32, PartitionsBin/binary>>) ->
     Partitions = list_of_strings(PartitionsBin),
diff --git a/deps/rabbitmq_stream_common/test/rabbit_stream_core_SUITE.erl b/deps/rabbitmq_stream_common/test/rabbit_stream_core_SUITE.erl
index 8e40b0f8d6..4766539c64 100644
--- a/deps/rabbitmq_stream_common/test/rabbit_stream_core_SUITE.erl
+++ b/deps/rabbitmq_stream_common/test/rabbit_stream_core_SUITE.erl
@@ -135,7 +135,8 @@ roundtrip(_Config) ->
     test_roundtrip({response, 0, {tune, 10000, 12345}}),
     %  %% NB: does not write correlation id
     test_roundtrip({response, 0, {credit, 98, 200}}),
-    test_roundtrip({response, 99, {route, 1, <<"stream_name">>}}),
+    test_roundtrip({response, 99,
+                    {route, 1, [<<"stream1">>, <<"stream2">>]}}),
     test_roundtrip({response, 99,
                     {partitions, 1, [<<"stream1">>, <<"stream2">>]}}),
     test_roundtrip({response, 99, {consumer_update, 1, none}}),