Simplify rabbit_queue_type callbacks

ansd · ansd · commit 5dd9ffc58255 · 2025-09-24T11:44:32.000+02:00
deliver should only take targets and init should only take the full record
diff --git a/deps/rabbit/src/rabbit_amqqueue.erl b/deps/rabbit/src/rabbit_amqqueue.erl
@@ -11,7 +11,7 @@
          delete_immediately/1, delete_exclusive/2, delete/4, purge/1,
          forget_all_durable/1]).
 -export([pseudo_queue/2, pseudo_queue/3]).
--export([exists/1, lookup/1, lookup/2, lookup_many/1, lookup_durable_queue/1,
+-export([exists/1, lookup/1, lookup/2, lookup_durable_queue/1,
          not_found_or_absent_dirty/1,
          with/2, with/3, with_or_die/2,
          assert_equivalence/5,
@@ -367,14 +367,6 @@ lookup(Name) when is_record(Name, resource) ->
 lookup_durable_queue(QName) ->
     rabbit_db_queue:get_durable(QName).
 
--spec lookup_many(rabbit_exchange:route_return()) ->
-    [amqqueue:amqqueue() | {amqqueue:amqqueue(), route_infos()}].
-lookup_many([]) ->
-    %% optimisation
-    [];
-lookup_many(Names) when is_list(Names) ->
-    rabbit_db_queue:get_many(Names).
-
 -spec lookup(binary(), binary()) ->
     rabbit_types:ok(amqqueue:amqqueue()) |
     rabbit_types:error('not_found').
diff --git a/deps/rabbit/src/rabbit_classic_queue.erl b/deps/rabbit/src/rabbit_classic_queue.erl
@@ -287,7 +287,7 @@ format(Q, _Ctx) when ?is_amqqueue(Q) ->
      {state, State},
      {node, node(amqqueue:get_pid(Q))}].
 
--spec init(amqqueue:amqqueue() | amqqueue:target()) ->
+-spec init(amqqueue:amqqueue()) ->
     {ok, state()}.
 init(Q) ->
     {ok, #?STATE{pid = amqqueue:get_pid(Q)}}.
@@ -474,10 +474,10 @@ settlement_action(Type, QRef, MsgSeqs, Acc) ->
 
 supports_stateful_delivery() -> true.
 
--spec deliver([{amqqueue:amqqueue() | amqqueue:target(), state()}],
+-spec deliver([{amqqueue:target(), state()}],
               Delivery :: mc:state(),
               rabbit_queue_type:delivery_options()) ->
-    {[{amqqueue:amqqueue(), state()}], rabbit_queue_type:actions()}.
+    {[{amqqueue:target(), state()}], rabbit_queue_type:actions()}.
 deliver(Qs0, Msg0, Options) ->
     %% add guid to content here instead of in rabbit_basic:message/3,
     %% as classic queues are the only ones that need it
diff --git a/deps/rabbit/src/rabbit_db_queue.erl b/deps/rabbit/src/rabbit_db_queue.erl
@@ -18,7 +18,6 @@
 
 -export([
          get/1,
-         get_many/1,
          get_targets/1,
          get_all/0,
          get_all/1,
@@ -474,106 +473,59 @@ internal_delete_in_mnesia(QueueName, OnlyDurable, Reason) ->
 %% get_targets().
 %% -------------------------------------------------------------------
 
-%% Queue target optimisation is only available in Khepri.
-%% Mnesia falls back looking up the full amqqueue record.
 -spec get_targets(rabbit_exchange:route_return()) ->
-    [amqqueue:target() | amqqueue:amqqueue() |
-     {amqqueue:target() | amqqueue:amqqueue(), rabbit_exchange:route_infos()}].
+    [amqqueue:target() | {amqqueue:target(), rabbit_exchange:route_infos()}].
 get_targets(Names) ->
     rabbit_khepri:handle_fallback(
-      #{mnesia => fun() -> get_many_in_ets(?MNESIA_TABLE, Names) end,
-        khepri => fun() -> lookup_targets(Names) end
+      #{mnesia => fun() -> lookup_targets(mnesia, Names) end,
+        khepri => fun() -> lookup_targets(khepri, Names) end
        }).
 
-lookup_targets(Names) ->
+lookup_targets(Store, Names) ->
     lists:filtermap(fun({Name, RouteInfos})
                           when is_map(RouteInfos) ->
-                            case lookup_target(Name) of
+                            case lookup_target(Store, Name) of
                                 not_found -> false;
                                 Target -> {true, {Target, RouteInfos}}
                             end;
                        (Name) ->
-                            case lookup_target(Name) of
+                            case lookup_target(Store, Name) of
                                 not_found -> false;
                                 Target -> {true, Target}
                             end
                     end, Names).
 
-lookup_target(#resource{name = NameBin} = Name) ->
+lookup_target(Store, #resource{name = NameBin} = Name) ->
     case rabbit_volatile_queue:is(NameBin) of
         true ->
-            %% This queue is not stored in the database.
-            %% We create it on the fly.
-            case rabbit_volatile_queue:new(Name) of
+            %% This queue is not stored in the database. We create it on the fly.
+            case rabbit_volatile_queue:new_target(Name) of
                 error -> not_found;
-                Q -> Q
+                Target -> Target
             end;
         false ->
-            try
-                ets:lookup_element(?KHEPRI_TARGET_PROJECTION, Name, 2, not_found) of
-                not_found ->
-                    not_found;
-                Target ->
-                    amqqueue:new_target(Name, Target)
-            catch
-                error:badarg ->
-                    not_found
-            end
+            lookup_target0(Store, Name)
     end.
 
-%% -------------------------------------------------------------------
-%% get_many().
-%% -------------------------------------------------------------------
-
--spec get_many(rabbit_exchange:route_return()) ->
-    [amqqueue:amqqueue() | {amqqueue:amqqueue(), rabbit_exchange:route_infos()}].
-get_many(Names) when is_list(Names) ->
-    rabbit_khepri:handle_fallback(
-      #{mnesia => fun() -> get_many_in_ets(?MNESIA_TABLE, Names) end,
-        khepri => fun() -> get_many_in_khepri(Names) end
-       }).
-
-get_many_in_khepri(Names) ->
-    try
-        get_many_in_ets(?KHEPRI_PROJECTION, Names)
+lookup_target0(khepri, Name) ->
+    try ets:lookup_element(?KHEPRI_TARGET_PROJECTION, Name, 2, not_found) of
+        not_found ->
+            not_found;
+        Target ->
+            amqqueue:new_target(Name, Target)
     catch
         error:badarg ->
-            []
-    end.
-
-get_many_in_ets(Table, [{Name, RouteInfos}])
-  when is_map(RouteInfos) ->
-    case ets_lookup(Table, Name) of
-        [] -> [];
-        [Q] -> [{Q, RouteInfos}]
+            not_found
     end;
-get_many_in_ets(Table, [Name]) ->
-    ets_lookup(Table, Name);
-get_many_in_ets(Table, Names) when is_list(Names) ->
-    lists:filtermap(fun({Name, RouteInfos})
-                          when is_map(RouteInfos) ->
-                            case ets_lookup(Table, Name) of
-                                [] -> false;
-                                [Q] -> {true, {Q, RouteInfos}}
-                            end;
-                       (Name) ->
-                            case ets_lookup(Table, Name) of
-                                [] -> false;
-                                [Q] -> {true, Q}
-                            end
-                    end, Names).
-
-ets_lookup(Table, QName = #resource{name = QNameBin}) ->
-    case rabbit_volatile_queue:is(QNameBin) of
-        true ->
-            %% This queue record is not stored in the database.
-            %% We create it on the fly.
-            case rabbit_volatile_queue:new(QName) of
-                error -> [];
-                Q -> [Q]
-            end;
-        false ->
-            ets:lookup(Table, QName)
+lookup_target0(mnesia, Name) ->
+    case ets:lookup(?MNESIA_TABLE, Name) of
+        [] ->
+            not_found;
+        [Q] ->
+            Type = amqqueue:get_type(Q),
+            Pid = amqqueue:get_pid(Q),
+            ExtraBcc = amqqueue:get_extra_bcc(Q),
+            amqqueue:new_target(Name, {Type, Pid, ExtraBcc})
     end.
 
 %% -------------------------------------------------------------------
@@ -663,6 +615,14 @@ get_many_durable_in_khepri(Names) ->
             []
     end.
 
+get_many_in_ets(Table, Names) ->
+    lists:filtermap(fun(Name) ->
+                            case ets:lookup(Table, Name) of
+                                [] -> false;
+                                [Q] -> {true, Q}
+                            end
+                    end, Names).
+
 %% -------------------------------------------------------------------
 %% update().
 %% -------------------------------------------------------------------
diff --git a/deps/rabbit/src/rabbit_queue_type.erl b/deps/rabbit/src/rabbit_queue_type.erl
@@ -204,7 +204,7 @@
 -callback policy_changed(amqqueue:amqqueue()) -> ok.
 
 %% intitialise and return a queue type specific session context
--callback init(amqqueue:amqqueue() | amqqueue:target()) ->
+-callback init(amqqueue:amqqueue()) ->
     {ok, queue_state()} | {error, Reason :: term()}.
 
 -callback close(queue_state()) -> ok.
@@ -233,10 +233,10 @@
 
 -callback supports_stateful_delivery() -> boolean().
 
--callback deliver([{amqqueue:amqqueue() | amqqueue:target(), queue_state()}],
+-callback deliver([{amqqueue:target(), queue_state()}],
                   Message :: mc:state(),
                   Options :: delivery_options()) ->
-    {[{amqqueue:amqqueue() | amqqueue:target(), queue_state()}], actions()}.
+    {[{amqqueue:target(), queue_state()}], actions()}.
 
 -callback settle(queue_name(), settle_op(), rabbit_types:ctag(),
                  [non_neg_integer()], queue_state()) ->
@@ -626,9 +626,8 @@ publish_at_most_once(X, Msg)
     _ = deliver(Qs, Msg, #{}, stateless),
     ok.
 
--spec deliver([amqqueue:amqqueue() | amqqueue:target() |
-               {amqqueue:amqqueue() | amqqueue:target(),
-                rabbit_exchange:route_infos()}],
+-spec deliver([amqqueue:target() |
+               {amqqueue:target(), rabbit_exchange:route_infos()}],
               Message :: mc:state(),
               delivery_options(),
               stateless | state()) ->
@@ -792,12 +791,14 @@ get_ctx_with(Q, #?STATE{ctxs = Contexts}, InitState) ->
         _ when InitState == undefined ->
             %% not found and no initial state passed - initialize new state
             Mod = amqqueue:get_type(Q),
-            case Mod:init(Q) of
+            maybe
+                {ok, Q1} ?= to_queue(Q),
+                {ok, QState} ?= Mod:init(Q1),
+                #ctx{module = Mod,
+                     state = QState}
+            else
                 {error, Reason} ->
-                    exit({Reason, Ref});
-                {ok, QState} ->
-                    #ctx{module = Mod,
-                         state = QState}
+                    exit({Reason, Ref})
             end;
         _  ->
             %% not found - initialize with supplied initial state
@@ -820,6 +821,12 @@ qref(#resource{kind = queue} = QName) ->
 qref(Q) ->
     amqqueue:get_name(Q).
 
+to_queue(Q) when ?is_amqqueue(Q) ->
+    {ok, Q};
+to_queue(Target) ->
+    QName = amqqueue:get_name(Target),
+    rabbit_amqqueue:lookup(QName).
+
 -spec known_queue_type_modules() -> [module()].
 known_queue_type_modules() ->
     Registered = rabbit_registry:lookup_all(queue),
diff --git a/deps/rabbit/src/rabbit_quorum_queue.erl b/deps/rabbit/src/rabbit_quorum_queue.erl
@@ -212,7 +212,7 @@ is_compatible(_Durable = true,
 is_compatible(_, _, _) ->
     false.
 
--spec init(amqqueue:amqqueue() | amqqueue:target()) ->
+-spec init(amqqueue:amqqueue()) ->
     {ok, rabbit_fifo_client:state()} | {error, not_found}.
 init(Q) when ?is_amqqueue(Q) ->
     {ok, SoftLimit} = application:get_env(rabbit, quorum_commands_soft_limit),
@@ -229,15 +229,7 @@ init(Q) when ?is_amqqueue(Q) ->
     %% server tried is the one we want
     Servers0 = [{Name, N} || N <- Nodes],
     Servers = [Leader | lists:delete(Leader, Servers0)],
-    {ok, rabbit_fifo_client:init(Servers, SoftLimit)};
-init(QueueTarget) ->
-    QName = amqqueue:get_name(QueueTarget),
-    case rabbit_amqqueue:lookup(QName) of
-        {ok, Q} ->
-            init(Q);
-        Error ->
-            Error
-    end.
+    {ok, rabbit_fifo_client:init(Servers, SoftLimit)}.
 
 -spec close(rabbit_fifo_client:state()) -> ok.
 close(State) ->
diff --git a/deps/rabbit/src/rabbit_stream_coordinator.erl b/deps/rabbit/src/rabbit_stream_coordinator.erl
@@ -169,20 +169,23 @@ restart_stream(QRes) ->
     {ok, node()} |
     {error, term()} |
     {timeout, term()}.
-restart_stream(QRes, Options)
-  when element(1, QRes) == resource ->
-    restart_stream(hd(rabbit_amqqueue:lookup_many([QRes])), Options);
 restart_stream(Q, Options)
-  when ?is_amqqueue(Q) andalso
-       ?amqqueue_is_stream(Q) ->
+  when ?amqqueue_is_stream(Q) ->
     ?LOG_INFO("restarting stream ~s in vhost ~s with options ~p",
-                    [maps:get(name, amqqueue:get_type_state(Q)), amqqueue:get_vhost(Q), Options]),
+              [maps:get(name, amqqueue:get_type_state(Q)), amqqueue:get_vhost(Q), Options]),
     #{name := StreamId} = amqqueue:get_type_state(Q),
     case process_command({restart_stream, StreamId, Options}) of
         {ok, {ok, LeaderPid}, _} ->
             {ok, node(LeaderPid)};
         Err ->
             Err
+    end;
+restart_stream(QRes, Options) ->
+    case rabbit_amqqueue:lookup(QRes) of
+        {ok, Q} ->
+            restart_stream(Q, Options);
+        Err ->
+            Err
     end.
 
 delete_stream(Q, ActingUser)
diff --git a/deps/rabbit/src/rabbit_stream_queue.erl b/deps/rabbit/src/rabbit_stream_queue.erl
@@ -1015,14 +1015,6 @@ init(Q) when ?is_amqqueue(Q) ->
             ?LOG_WARNING("Failed to start stream client ~tp: coordinator unavailable",
                          [rabbit_misc:rs(QName)]),
             E
-    end;
-init(QueueTarget) ->
-    QName = amqqueue:get_name(QueueTarget),
-    case rabbit_amqqueue:lookup(QName) of
-        {ok, Q} ->
-            init(Q);
-        Error ->
-            Error
     end.
 
 close(#stream_client{readers = Readers,
diff --git a/deps/rabbit/src/rabbit_volatile_queue.erl b/deps/rabbit/src/rabbit_volatile_queue.erl
@@ -16,6 +16,7 @@
 -include_lib("rabbit_common/include/rabbit.hrl").
 
 -export([new/1,
+         new_target/1,
          new_name/0,
          is/1,
          key_from_name/1,
@@ -94,6 +95,16 @@ new(#resource{virtual_host = Vhost,
 new0(Name, Pid, Vhost) ->
     amqqueue:new(Name, Pid, false, true, none, [], Vhost, #{}, ?MODULE).
 
+-spec new_target(rabbit_amqqueue:name()) ->
+    amqqueue:target() | error.
+new_target(#resource{name = NameBin} = Name) ->
+    case pid_from_name(NameBin) of
+        {ok, Pid} when is_pid(Pid) ->
+            amqqueue:new_target(Name, {?MODULE, Pid, none});
+        _ ->
+            error
+    end.
+
 -spec is(rabbit_misc:resource_name()) ->
     boolean().
 is(<<?PREFIX, _/binary>>) ->
diff --git a/deps/rabbit/test/rabbit_db_queue_SUITE.erl b/deps/rabbit/test/rabbit_db_queue_SUITE.erl
diff --git a/deps/rabbitmq_mqtt/src/rabbit_mqtt_qos0_queue.erl b/deps/rabbitmq_mqtt/src/rabbit_mqtt_qos0_queue.erl
