MOve rabbit_routing_util to STOMP #13009 #13016

This avoids making `amqp_client` aware of the DQT
code paths in the server, plus its Dialyzer run
now passes.

(cherry picked from commit b2d5376)

Author: michaelklishin

deps/rabbit/src/rabbit_amqqueue.erl

@@ -248,13 +248,26 @@ declare(QueueName = #resource{virtual_host = VHost}, Durable, AutoDelete, Args,
 
 -spec get_queue_type(Args :: rabbit_framing:amqp_table()) -> rabbit_queue_type:queue_type().
 %% This version is not virtual host metadata-aware.
+get_queue_type([]) ->
+    rabbit_queue_type:default();
 get_queue_type(Args) ->
-    rabbit_queue_type_common:get_queue_type(Args, rabbit_queue_type:default()).
+    get_queue_type(Args, rabbit_queue_type:default()).
 
 -spec get_queue_type(Args :: rabbit_framing:amqp_table(), DefaultQueueType :: rabbit_queue_type:queue_type()) -> rabbit_queue_type:queue_type().
 %% This version should be used together with 'rabbit_vhost:default_queue_type/{1,2}'.
+get_queue_type([], DefaultQueueType) ->
+    rabbit_queue_type:discover(DefaultQueueType);
 get_queue_type(Args, DefaultQueueType) ->
-    rabbit_queue_type_common:get_queue_type(Args, DefaultQueueType).
+    case rabbit_misc:table_lookup(Args, <<"x-queue-type">>) of
+        undefined ->
+            rabbit_queue_type:discover(DefaultQueueType);
+        {longstr, undefined} ->
+            rabbit_queue_type:discover(DefaultQueueType);
+        {longstr, <<"undefined">>} ->
+            rabbit_queue_type:discover(DefaultQueueType);
+        {_, V} ->
+            rabbit_queue_type:discover(V)
+    end.
 
 -spec internal_declare(Queue, Recover) -> Ret when
       Queue :: amqqueue:amqqueue(),

deps/rabbit_common/src/rabbit_queue_type_common.erl

@@ -2,7 +2,7 @@
 %% License, v. 2.0. If a copy of the MPL was not distributed with this
 %% file, You can obtain one at https://mozilla.org/MPL/2.0/.
 %%
-%% Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+%% Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
 %%
 
 -define(QUEUE_PREFIX, "/queue").

…ient/include/rabbit_routing_prefixes.hrl …nclude/rabbit_stomp_routing_prefixes.hrldeps/amqp_client/include/rabbit_routing_prefixes.hrl renamed to deps/rabbitmq_stomp/include/rabbit_stomp_routing_prefixes.hrl deps/amqp_client/include/rabbit_routing_prefixes.hrl renamed to deps/rabbitmq_stomp/include/rabbit_stomp_routing_prefixes.hrl

@@ -153,7 +153,7 @@ initial_state(Configuration,
        version             = none,
        pending_receipts    = undefined,
        config              = Configuration,
-       route_state         = rabbit_routing_util:init_state(),
+       route_state         = rabbit_stomp_routing_util:init_state(),
        reply_queues        = #{},
        frame_transformer   = undefined,
        trailing_lf         = application:get_env(rabbitmq_stomp, trailing_lf, true),
@@ -557,7 +557,7 @@ with_destination(Command, Frame, State, Fun) ->
                           "'~ts' is not a valid destination.~n"
                           "Valid destination types are: ~ts.~n",
                           [Content,
-                           string:join(rabbit_routing_util:all_dest_prefixes(),
+                           string:join(rabbit_stomp_routing_util:all_dest_prefixes(),
                                        ", ")], State)
             end;
         not_found ->
@@ -652,15 +652,14 @@ server_header() ->
 
 do_subscribe(Destination, DestHdr, Frame,
              State = #proc_state{subscriptions = Subs,
-                                 route_state   = RouteState,
                                  channel       = Channel,
                                  default_topic_exchange = DfltTopicEx}) ->
     check_subscription_access(Destination, State),
     Prefetch =
         rabbit_stomp_frame:integer_header(Frame, ?HEADER_PREFETCH_COUNT,
                                           undefined),
     {AckMode, IsMulti} = rabbit_stomp_util:ack_mode(Frame),
-    case ensure_endpoint(source, Destination, Frame, Channel, RouteState) of
+    case ensure_endpoint(source, Destination, Frame, Channel, State) of
         {ok, Queue, RouteState1} ->
             {ok, ConsumerTag, Description} =
                 rabbit_stomp_util:consumer_tag(Frame),
@@ -689,7 +688,7 @@ do_subscribe(Destination, DestHdr, Frame,
                                                   exclusive    = false,
                                                   arguments    = Arguments},
                                                self()),
-                        ok = rabbit_routing_util:ensure_binding(
+                        ok = rabbit_stomp_routing_util:ensure_binding(
                                Queue, ExchangeAndKey, Channel)
                     catch exit:Err ->
                             %% it's safe to delete this queue, it
@@ -790,9 +789,8 @@ maybe_clean_up_queue(Queue, #proc_state{connection = Connection}) ->
 do_send(Destination, _DestHdr,
         Frame = #stomp_frame{body_iolist = BodyFragments},
         State = #proc_state{channel = Channel,
-                            route_state = RouteState,
                             default_topic_exchange = DfltTopicEx}) ->
-    case ensure_endpoint(dest, Destination, Frame, Channel, RouteState) of
+    case ensure_endpoint(dest, Destination, Frame, Channel, State) of
 
         {ok, _Q, RouteState1} ->
 
@@ -914,7 +912,7 @@ ensure_reply_to(Frame = #stomp_frame{headers = Headers}, State) ->
             {Frame, State};
         {ok, ReplyTo} ->
             {ok, Destination} = rabbit_routing_parser:parse_endpoint(ReplyTo),
-            case rabbit_routing_util:dest_temp_queue(Destination) of
+            case rabbit_stomp_routing_util:dest_temp_queue(Destination) of
                 none ->
                     {Frame, State};
                 TempQueueId ->
@@ -1131,7 +1129,7 @@ ensure_endpoint(_Direction, {queue, []}, _Frame, _Channel, _State) ->
     {error, {invalid_destination, "Destination cannot be blank"}};
 
 ensure_endpoint(source, EndPoint, {_, _, Headers, _} = Frame, Channel,
-                State = #proc_state{virtual_host = VHost}) ->
+                #proc_state{virtual_host = VHost, route_state = RouteState}) ->
     Params =
         [{subscription_queue_name_gen,
           fun () ->
@@ -1145,16 +1143,16 @@ ensure_endpoint(source, EndPoint, {_, _, Headers, _} = Frame, Channel,
          {default_queue_type, rabbit_vhost:default_queue_type(VHost)}]
         ++ rabbit_stomp_util:build_params(EndPoint, Headers),
     Arguments = rabbit_stomp_util:build_arguments(Headers),
-    rabbit_routing_util:ensure_endpoint(source, Channel, EndPoint,
-                                        [Arguments | Params], State);
+    rabbit_stomp_routing_util:ensure_endpoint(source, Channel, EndPoint,
+                                        [Arguments | Params], RouteState);
 
 ensure_endpoint(Direction, EndPoint, {_, _, Headers, _}, Channel,
-                State = #proc_state{virtual_host = VHost}) ->
+                #proc_state{virtual_host = VHost, route_state = RouteState}) ->
     Params = [{default_queue_type, rabbit_vhost:default_queue_type(VHost)}
               | rabbit_stomp_util:build_params(EndPoint, Headers)],
     Arguments = rabbit_stomp_util:build_arguments(Headers),
-    rabbit_routing_util:ensure_endpoint(Direction, Channel, EndPoint,
-                                        [Arguments | Params], State).
+    rabbit_stomp_routing_util:ensure_endpoint(Direction, Channel, EndPoint,
+                                        [Arguments | Params], RouteState).
 
 build_subscription_id(Frame) ->
     case rabbit_stomp_util:has_durable_header(Frame) of

deps/rabbitmq_stomp/src/rabbit_stomp_processor.erl

@@ -2,17 +2,17 @@
 %% License, v. 2.0. If a copy of the MPL was not distributed with this
 %% file, You can obtain one at https://mozilla.org/MPL/2.0/.
 %%
-%% Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+%% Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
 %%
 
--module(rabbit_routing_util).
+-module(rabbit_stomp_routing_util).
 
 -export([init_state/0, dest_prefixes/0, all_dest_prefixes/0]).
 -export([ensure_endpoint/4, ensure_endpoint/5, ensure_binding/3]).
 -export([dest_temp_queue/1]).
 
--include("amqp_client.hrl").
--include("rabbit_routing_prefixes.hrl").
+-include_lib("amqp_client/include/amqp_client.hrl").
+-include("rabbit_stomp_routing_prefixes.hrl").
 
 %%----------------------------------------------------------------------------
 
@@ -147,8 +147,8 @@ queue_declare_method(#'queue.declare'{} = Method, Type, Params) ->
 
     Arguments = proplists:get_value(arguments, Params, []),
     DefaultQueueType = proplists:get_value(default_queue_type, Params,
-                                           rabbit_queue_type_common:default()),
-    Method3 = case rabbit_queue_type_common:get_queue_type(Arguments, DefaultQueueType) of
+                                           rabbit_queue_type:default()),
+    Method3 = case rabbit_amqqueue:get_queue_type(Arguments, DefaultQueueType) of
                   T when T =:= rabbit_stream_queue; T =:= rabbit_quorum_queue ->
                       Method2#'queue.declare'{durable   = true,
                                              exclusive = false};

…/amqp_client/src/rabbit_routing_util.erl …_stomp/src/rabbit_stomp_routing_util.erldeps/amqp_client/src/rabbit_routing_util.erl renamed to deps/rabbitmq_stomp/src/rabbit_stomp_routing_util.erl deps/amqp_client/src/rabbit_routing_util.erl renamed to deps/rabbitmq_stomp/src/rabbit_stomp_routing_util.erl

@@ -17,7 +17,7 @@
 -export([trim_headers/1]).
 
 -include_lib("amqp_client/include/amqp_client.hrl").
--include_lib("amqp_client/include/rabbit_routing_prefixes.hrl").
+-include("rabbit_stomp_routing_prefixes.hrl").
 -include("rabbit_stomp_frame.hrl").
 -include("rabbit_stomp_headers.hrl").
 
@@ -135,7 +135,7 @@ headers_extra(SessionId, AckMode, Version,
     end.
 
 headers_post_process(Headers) ->
-    Prefixes = rabbit_routing_util:dest_prefixes(),
+    Prefixes = rabbit_stomp_routing_util:dest_prefixes(),
     [case Header of
          {?HEADER_REPLY_TO, V} ->
              case lists:any(fun (P) -> lists:prefix(P, V) end, Prefixes) of
@@ -333,7 +333,7 @@ default_params({exchange, _}) ->
     [{exclusive, true}, {auto_delete, true}];
 
 default_params({topic, _}) ->
-    [{exclusive, false}, {auto_delete, true}];
+    [{auto_delete, true}];
 
 default_params(_) ->
     [{exclusive, true},

deps/rabbitmq_stomp/src/rabbit_stomp_util.erl

@@ -46,16 +46,9 @@ init_per_group(_, Config) ->
                                                {rmq_certspwd, "bunnychow"}
                                             ]),
     rabbit_ct_helpers:log_environment(),
-    %% Remove when transient_nonexcl_queues is removed entirely,
-    %% and fix the Python tests.
-    Config1 = rabbit_ct_helpers:merge_app_env(
-                Config0,
-                {rabbit,
-                 [{permit_deprecated_features, #{transient_nonexcl_queues => true}}]}),
-    Config2 = rabbit_ct_helpers:run_setup_steps(
-        Config1,
-        rabbit_ct_broker_helpers:setup_steps()),
-    Config2.
+    rabbit_ct_helpers:run_setup_steps(
+        Config0,
+        rabbit_ct_broker_helpers:setup_steps()).
 
 end_per_group(_, Config) ->
     rabbit_ct_helpers:run_teardown_steps(Config,

deps/rabbitmq_stomp/test/python_SUITE.erl

@@ -9,7 +9,7 @@
 
 -include_lib("eunit/include/eunit.hrl").
 -include_lib("amqp_client/include/amqp_client.hrl").
--include_lib("amqp_client/include/rabbit_routing_prefixes.hrl").
+-include("rabbit_stomp_routing_prefixes.hrl").
 -include("rabbit_stomp_frame.hrl").
 -compile(export_all).
 
@@ -149,7 +149,7 @@ headers_post_process_noop_replyto(_) ->
     [begin
          Headers = [{"reply-to", Prefix ++ "/something"}],
          Headers = rabbit_stomp_util:headers_post_process(Headers)
-     end || Prefix <- rabbit_routing_util:dest_prefixes()].
+     end || Prefix <- rabbit_stomp_routing_util:dest_prefixes()].
 
 headers_post_process_noop2(_) ->
     Headers  = [{"header1", "1"},