Move record queue_target into module amqqueue

Author: ansd

deps/rabbit/src/amqqueue.erl

@@ -14,6 +14,7 @@
          new/9,
          new_with_version/9,
          new_with_version/10,
+         to_target/1,
          fields/0,
          fields/1,
          field_vhost/0,
@@ -56,6 +57,7 @@
          set_state/2,
          get_type/1,
          get_vhost/1,
+         get_extra_bcc/1,
          is_amqqueue/1,
          is_auto_delete/1,
          is_durable/1,
@@ -120,6 +122,17 @@
           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.
+-record(queue_target,
+        {name :: rabbit_amqqueue:name(),
+         type :: rabbit_queue_type:queue_type(),
+         pid :: pid() | ra_server_id() | none,
+         extra_bcc :: rabbit_misc:resource_name() | none
+        }).
+
+-type target() :: #queue_target{}.
+
 -type amqqueue() :: amqqueue_v2().
 -type amqqueue_v2() :: #amqqueue{
                           name :: rabbit_amqqueue:name(),
@@ -145,8 +158,6 @@
                           type_state :: #{}
                          }.
 
--type target() :: #queue_target{}.
-
 -type ra_server_id() :: {Name :: atom(), Node :: node()}.
 
 -type amqqueue_pattern() :: amqqueue_v2_pattern().
@@ -331,6 +342,26 @@ 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}) ->
+    #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.
+
 -spec is_amqqueue(any()) -> boolean().
 
 is_amqqueue(#amqqueue{}) -> true.

deps/rabbit/src/rabbit_amqqueue.erl

@@ -2057,7 +2057,7 @@ prepend_extra_bcc([]) ->
     [];
 prepend_extra_bcc([Q0] = Qs) ->
     Q = queue(Q0),
-    case get_extra_bcc(Q) of
+    case amqqueue:get_extra_bcc(Q) of
         none ->
             Qs;
         Name ->
@@ -2072,7 +2072,7 @@ prepend_extra_bcc(Qs) ->
     ExtraQs = lists:filtermap(
                 fun(Q0) ->
                         Q = queue(Q0),
-                        case get_extra_bcc(Q) of
+                        case amqqueue:get_extra_bcc(Q) of
                             none ->
                                 false;
                             Name ->
@@ -2102,16 +2102,6 @@ queue_names(Queues) ->
                       amqqueue:get_name(Q)
               end, Queues).
 
-get_extra_bcc(Q) when ?is_amqqueue(Q) ->
-    case amqqueue:get_options(Q) of
-        #{extra_bcc := Name} ->
-            Name;
-        _ ->
-            none
-    end;
-get_extra_bcc(#queue_target{extra_bcc = Name}) ->
-    Name.
-
 -spec lookup_extra_bcc(amqqueue:amqqueue() | amqqueue:target(), binary()) ->
     {ok, amqqueue:amqqueue()} | {error, not_found}.
 lookup_extra_bcc(Q, BCCName) ->

deps/rabbit/src/rabbit_khepri.erl

@@ -1358,17 +1358,8 @@ register_rabbit_queue_target_projection() ->
     PathPattern = rabbit_db_queue:khepri_queue_path(
                     _VHost = ?KHEPRI_WILDCARD_STAR,
                     _Name = ?KHEPRI_WILDCARD_STAR),
-    Fun = fun(_Path, Q) ->
-                  BCC = case amqqueue:get_options(Q) of
-                            #{extra_bcc := Name} -> Name;
-                            _ -> none
-                        end,
-                  #queue_target{name = amqqueue:get_name(Q),
-                                type = amqqueue:get_type(Q),
-                                pid = amqqueue:get_pid(Q),
-                                extra_bcc = BCC}
-          end,
-    Opts = #{keypos => #queue_target.name,
+    Fun = fun(_Path, Q) -> amqqueue:to_target(Q) end,
+    Opts = #{keypos => 2, %% #queue_target.name
              read_concurrency => true},
     Projection = khepri_projection:new(rabbit_khepri_queue_target, Fun, Opts),
     khepri:register_projection(?STORE_ID, PathPattern, Projection).

deps/rabbit_common/include/rabbit.hrl

@@ -96,15 +96,6 @@
 -record(index_route, {source_key, destination, args = []}).
 -record(route_by_source, {source, key, destination, args = []}).
 
-%% A subset of the amqqueue record to avoid looking up the full amqqueue record
-%% when delivering a message to a target queue.
--record(queue_target,
-        {name :: tuple(),
-         type :: module(),
-         pid :: pid() | {atom(), node()} | none,
-         extra_bcc :: binary() | none
-        }).
-
 -record(binding, {source, key, destination, args = []}).
 -record(reverse_binding, {destination, key, source, args = []}).