Avoid copying the queue name

Author: ansd

deps/rabbit/src/amqqueue.erl

@@ -14,7 +14,7 @@
          new/9,
          new_with_version/9,
          new_with_version/10,
-         to_target/1,
+         new_target/4,
          fields/0,
          fields/1,
          field_vhost/0,
@@ -40,6 +40,7 @@
          % options
          get_options/1,
          set_options/2,
+         get_extra_bcc/1,
          % pid
          get_pid/1,
          set_pid/2,
@@ -57,7 +58,6 @@
          set_state/2,
          get_type/1,
          get_vhost/1,
-         get_extra_bcc/1,
          is_amqqueue/1,
          is_auto_delete/1,
          is_durable/1,
@@ -79,7 +79,8 @@
          qnode/1,
          to_printable/1,
          to_printable/2,
-         macros/0]).
+         macros/0
+        ]).
 
 -define(record_version, amqqueue_v2).
 -define(is_backwards_compat_classic(T),
@@ -122,8 +123,8 @@
           type_state = #{} :: map() | ets:match_pattern()
          }).
 
-%% A subset of the amqqueue record to avoid looking up the full amqqueue record
-%% when delivering a message to a target queue.
+%% A subset of the amqqueue record containing just the necessary fields
+%% to deliver a message to a target queue.
 -record(queue_target,
         {name :: rabbit_amqqueue:name(),
          type :: rabbit_queue_type:queue_type(),
@@ -342,25 +343,16 @@ new_with_version(?record_version,
               options         = Options,
               type            = ensure_type_compat(Type)}.
 
--spec to_target(amqqueue()) -> target().
-to_target(#amqqueue{name = Name,
-                    type = Type,
-                    pid = Pid,
-                    options = Options}) ->
+-spec new_target(rabbit_amqqueue:name(),
+                 rabbit_queue_type:queue_type(),
+                 pid() | ra_server_id() | none,
+                 rabbit_misc:resource_name() | none) ->
+    target().
+new_target(Name, Type, Pid, ExtraBcc) ->
     #queue_target{name = Name,
                   type = Type,
                   pid = Pid,
-                  extra_bcc = extra_bcc_from_options(Options)}.
-
-get_extra_bcc(#amqqueue{options = Options})  ->
-    extra_bcc_from_options(Options);
-get_extra_bcc(#queue_target{extra_bcc = Name}) ->
-    Name.
-
-extra_bcc_from_options(#{extra_bcc := Name}) ->
-    Name;
-extra_bcc_from_options(#{}) ->
-    none.
+                  extra_bcc = ExtraBcc}.
 
 -spec is_amqqueue(any()) -> boolean().
 
@@ -395,15 +387,21 @@ set_arguments(#amqqueue{} = Queue, Args) ->
 % options
 
 -spec get_options(amqqueue()) -> amqqueue_options().
-
 get_options(#amqqueue{options = Options}) ->
     Options.
 
 -spec set_options(amqqueue(), amqqueue_options()) -> amqqueue().
-
 set_options(#amqqueue{} = Queue, Options) ->
     Queue#amqqueue{options = Options}.
 
+-spec get_extra_bcc(amqqueue() | target()) ->
+    rabbit_misc:resource_name() | none.
+get_extra_bcc(#amqqueue{options = #{extra_bcc := Name}}) ->
+    Name;
+get_extra_bcc(#amqqueue{})  ->
+    none;
+get_extra_bcc(#queue_target{extra_bcc = Name}) ->
+    Name.
 
 % decorators
 
@@ -666,8 +664,6 @@ to_printable(QName = #resource{name = Name, virtual_host = VHost}, Type) ->
        <<"virtual_host">> => VHost,
        <<"type">> => Type}.
 
-% private
-
 macros() ->
     io:format(
       "-define(is_~ts(Q), is_record(Q, amqqueue, ~b)).~n~n",

deps/rabbit/src/rabbit_db_queue.erl

@@ -482,9 +482,45 @@ internal_delete_in_mnesia(QueueName, OnlyDurable, Reason) ->
 get_targets(Names) ->
     rabbit_khepri:handle_fallback(
       #{mnesia => fun() -> get_many_in_ets(?MNESIA_TABLE, Names) end,
-        khepri => fun() -> get_many_in_khepri(?KHEPRI_TARGET_PROJECTION, Names) end
+        khepri => fun() -> lookup_targets(Names) end
        }).
 
+lookup_targets(Names) ->
+    lists:filtermap(fun({Name, RouteInfos})
+                          when is_map(RouteInfos) ->
+                            case lookup_target(Name) of
+                                not_found -> false;
+                                Target -> {true, {Target, RouteInfos}}
+                            end;
+                       (Name) ->
+                            case lookup_target(Name) of
+                                not_found -> false;
+                                Target -> {true, Target}
+                            end
+                    end, Names).
+
+lookup_target(#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
+                error -> not_found;
+                Q -> Q
+            end;
+        false ->
+            try
+                ets:lookup_element(?KHEPRI_TARGET_PROJECTION, Name, 2, not_found) of
+                {Type, Pid, ExtraBcc} ->
+                    amqqueue:new_target(Name, Type, Pid, ExtraBcc);
+                not_found ->
+                    not_found
+            catch
+                error:badarg ->
+                    not_found
+            end
+    end.
+
 %% -------------------------------------------------------------------
 %% get_many().
 %% -------------------------------------------------------------------
@@ -494,12 +530,12 @@ get_targets(Names) ->
 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(?KHEPRI_PROJECTION, Names) end
+        khepri => fun() -> get_many_in_khepri(Names) end
        }).
 
-get_many_in_khepri(Table, Names) ->
+get_many_in_khepri(Names) ->
     try
-        get_many_in_ets(Table, Names)
+        get_many_in_ets(?KHEPRI_PROJECTION, Names)
     catch
         error:badarg ->
             []

deps/rabbit/src/rabbit_khepri.erl

@@ -1354,12 +1354,20 @@ register_rabbit_queue_projection() ->
     KeyPos = 2, %% #amqqueue.name
     register_simple_projection(Name, PathPattern, KeyPos, false).
 
+%% This projection exists to avoid looking up the full amqqueue record
+%% per message delivered to a target queue.
 register_rabbit_queue_target_projection() ->
     PathPattern = rabbit_db_queue:khepri_queue_path(
                     _VHost = ?KHEPRI_WILDCARD_STAR,
                     _Name = ?KHEPRI_WILDCARD_STAR),
-    Fun = fun(_Path, Q) -> amqqueue:to_target(Q) end,
-    Opts = #{keypos => 2, %% #queue_target.name
+    Fun = fun(_Path, Q) ->
+                  Name = amqqueue:get_name(Q),
+                  Type = amqqueue:get_type(Q),
+                  Pid = amqqueue:get_pid(Q),
+                  ExtraBcc = amqqueue:get_extra_bcc(Q),
+                  {Name, {Type, Pid, ExtraBcc}}
+          end,
+    Opts = #{keypos => 1,
              read_concurrency => true},
     Projection = khepri_projection:new(rabbit_khepri_queue_target, Fun, Opts),
     khepri:register_projection(?STORE_ID, PathPattern, Projection).