Enable adding queues with plugins, core

Author: deadtrickster

deps/rabbit/include/amqqueue.hrl

@@ -48,6 +48,10 @@
         (?is_amqqueue_v2(Q) andalso
          ?amqqueue_v2_field_type(Q) =:= Type)).
 
+-define(amqqueue_type(Q),
+        (?is_amqqueue_v2(Q) andalso
+         ?amqqueue_v2_field_type(Q))).
+
 -define(amqqueue_has_valid_pid(Q),
         (?is_amqqueue_v2(Q) andalso
          is_pid(?amqqueue_v2_field_pid(Q)))).

deps/rabbit/src/amqqueue.erl

@@ -368,6 +368,7 @@ get_exclusive_owner(#amqqueue{exclusive_owner = Owner}) ->
 
 -spec get_leader(amqqueue_v2()) -> node().
 
+%% TODO: not only qqs can have leaders, dispatch via queue type
 get_leader(#amqqueue{type = rabbit_quorum_queue, pid = {_, Leader}}) -> Leader.
 
 % operator_policy

deps/rabbit/src/rabbit_amqp_management.erl

@@ -461,32 +461,8 @@ encode_queue(Q, NumMsgs, NumConsumers) ->
 -spec queue_topology(amqqueue:amqqueue()) ->
     {Leader :: undefined | node(), Replicas :: undefined | [node(),...]}.
 queue_topology(Q) ->
-    case amqqueue:get_type(Q) of
-        rabbit_quorum_queue ->
-            [{leader, Leader0},
-             {members, Members}] = rabbit_queue_type:info(Q, [leader, members]),
-            Leader = case Leader0 of
-                         '' -> undefined;
-                         _ -> Leader0
-                     end,
-            {Leader, Members};
-        rabbit_stream_queue ->
-            #{name := StreamId} = amqqueue:get_type_state(Q),
-            case rabbit_stream_coordinator:members(StreamId) of
-                {ok, Members} ->
-                    maps:fold(fun(Node, {_Pid, writer}, {_, Replicas}) ->
-                                      {Node, [Node | Replicas]};
-                                 (Node, {_Pid, replica}, {Writer, Replicas}) ->
-                                      {Writer, [Node | Replicas]}
-                              end, {undefined, []}, Members);
-                {error, _} ->
-                    {undefined, undefined}
-            end;
-        _ ->
-            Pid = amqqueue:get_pid(Q),
-            Node = node(Pid),
-            {Node, [Node]}
-    end.
+    Type = amqqueue:get_type(Q),
+    Type:queue_topology(Q).
 
 decode_exchange({map, KVList}) ->
     M = lists:foldl(

deps/rabbit/src/rabbit_amqqueue.erl

@@ -150,11 +150,7 @@ filter_pid_per_type(QPids) ->
 
 -spec stop(rabbit_types:vhost()) -> 'ok'.
 stop(VHost) ->
-    %% Classic queues
-    ok = rabbit_amqqueue_sup_sup:stop_for_vhost(VHost),
-    {ok, BQ} = application:get_env(rabbit, backing_queue_module),
-    ok = BQ:stop(VHost),
-    rabbit_quorum_queue:stop(VHost).
+    rabbit_queue_type:stop(VHost).
 
 -spec start([amqqueue:amqqueue()]) -> 'ok'.
 
@@ -424,6 +420,8 @@ rebalance(Type, VhostSpec, QueueSpec) ->
     %% We have not yet acquired the rebalance_queues global lock.
     maybe_rebalance(get_rebalance_lock(self()), Type, VhostSpec, QueueSpec).
 
+%% TODO: classic queues do not support rebalancing, it looks like they are simply
+%% filtered out with is_replicated(Q). Maybe error instead?
 maybe_rebalance({true, Id}, Type, VhostSpec, QueueSpec) ->
     rabbit_log:info("Starting queue rebalance operation: '~ts' for vhosts matching '~ts' and queues matching '~ts'",
                     [Type, VhostSpec, QueueSpec]),
@@ -459,10 +457,15 @@ filter_per_type(stream, Q) ->
 filter_per_type(classic, Q) ->
     ?amqqueue_is_classic(Q).
 
-rebalance_module(Q) when ?amqqueue_is_quorum(Q) ->
-    rabbit_quorum_queue;
-rebalance_module(Q) when ?amqqueue_is_stream(Q) ->
-    rabbit_stream_queue.
+%% TODO: note that it can return {error, not_supported}.
+%% this will result in a badmatch. However that's fine
+%% for now because the original function will fail with
+%% bad clause if called with classical queue.
+%% The assumption is all non-replicated queues
+%% are filtered before calling this with is_replicated/0
+rebalance_module(Q) ->
+    TypeModule = ?amqqueue_type(Q),
+    TypeModule:rebalance_module().
 
 get_resource_name(#resource{name = Name}) ->
     Name.
@@ -487,13 +490,19 @@ iterative_rebalance(ByNode, MaxQueuesDesired) ->
 maybe_migrate(ByNode, MaxQueuesDesired) ->
     maybe_migrate(ByNode, MaxQueuesDesired, maps:keys(ByNode)).
 
+%% TODO: unfortunate part - UI bits mixed deep inside logic.
+%% I will not be moving this inside queue type. Instead
+%% an attempt to generate something more readable than
+%% Other made.
 column_name(rabbit_classic_queue) -> <<"Number of replicated classic queues">>;
 column_name(rabbit_quorum_queue) -> <<"Number of quorum queues">>;
 column_name(rabbit_stream_queue) -> <<"Number of streams">>;
-column_name(Other) -> Other.
+column_name(TypeModule) ->
+    Alias = rabbit_queue_type:short_alias_of(TypeModule),
+    <<"Number of \"", Alias/binary, "\" queues">>.
 
 maybe_migrate(ByNode, _, []) ->
-    ByNodeAndType = maps:map(fun(_Node, Queues) -> maps:groups_from_list(fun({_, Q, _}) -> column_name(?amqqueue_v2_field_type(Q)) end, Queues) end, ByNode),
+    ByNodeAndType = maps:map(fun(_Node, Queues) -> maps:groups_from_list(fun({_, Q, _}) -> column_name(?amqqueue_type(Q)) end, Queues) end, ByNode),
     CountByNodeAndType = maps:map(fun(_Node, Type) -> maps:map(fun (_, Qs)-> length(Qs) end, Type) end, ByNodeAndType),
     {ok, maps:values(maps:map(fun(Node,Counts) -> [{<<"Node name">>, Node} | maps:to_list(Counts)] end, CountByNodeAndType))};
 maybe_migrate(ByNode, MaxQueuesDesired, [N | Nodes]) ->
@@ -1252,14 +1261,12 @@ list_durable() ->
 
 -spec list_by_type(atom()) -> [amqqueue:amqqueue()].
 
-list_by_type(classic) -> list_by_type(rabbit_classic_queue);
-list_by_type(quorum)  -> list_by_type(rabbit_quorum_queue);
-list_by_type(stream)  -> list_by_type(rabbit_stream_queue);
-list_by_type(Type) ->
-    rabbit_db_queue:get_all_durable_by_type(Type).
+list_by_type(TypeDescriptor) ->
+    TypeModule = rabbit_queue_type:discover(TypeDescriptor),
+    rabbit_db_queue:get_all_durable_by_type(TypeModule).
 
+%% TODO: looks unused
 -spec list_local_quorum_queue_names() -> [name()].
-
 list_local_quorum_queue_names() ->
     [ amqqueue:get_name(Q) || Q <- list_by_type(quorum),
            amqqueue:get_state(Q) =/= crashed,
@@ -1296,6 +1303,7 @@ list_local_followers() ->
          rabbit_quorum_queue:is_recoverable(Q)
          ].
 
+%% TODO: looks unused
 -spec list_local_quorum_queues_with_name_matching(binary()) -> [amqqueue:amqqueue()].
 list_local_quorum_queues_with_name_matching(Pattern) ->
     [ Q || Q <- list_by_type(quorum),
@@ -1882,11 +1890,9 @@ run_backing_queue(QPid, Mod, Fun) ->
 
 -spec is_replicated(amqqueue:amqqueue()) -> boolean().
 
-is_replicated(Q) when ?amqqueue_is_classic(Q) ->
-    false;
-is_replicated(_Q) ->
-    %% streams and quorum queues are all replicated
-    true.
+is_replicated(Q) ->
+    TypeModule = ?amqqueue_type(Q),
+    TypeModule:is_replicated().
 
 is_exclusive(Q) when ?amqqueue_exclusive_owner_is(Q, none) ->
     false;

deps/rabbit/src/rabbit_boot_steps.erl

@@ -4,6 +4,8 @@
 %%
 %% Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
 %%
+%% README: https://github.com/rabbitmq/internals/blob/master/rabbit_boot_process.md
+%%
 
 -module(rabbit_boot_steps).
 

deps/rabbit/src/rabbit_channel.erl

@@ -1265,11 +1265,7 @@ handle_method(#'basic.get'{queue = QueueNameBin, no_ack = NoAck},
             ?INCR_STATS(queue_stats, QueueName, 1, get_empty, State),
             {reply, #'basic.get_empty'{}, State#ch{queue_states = QueueStates}};
         {error, {unsupported, single_active_consumer}} ->
-            rabbit_misc:protocol_error(
-              resource_locked,
-              "cannot obtain access to locked ~ts. basic.get operations "
-              "are not supported by quorum queues with single active consumer",
-              [rabbit_misc:rs(QueueName)]);
+             rabbit_amqqueue:with_or_die(QueueName, fun unsupported_single_active_consumer_error/1);
         {error, Reason} ->
             %% TODO add queue type to error message
             rabbit_misc:protocol_error(internal_error,
@@ -2005,6 +2001,7 @@ foreach_per_queue(_F, [], Acc) ->
 foreach_per_queue(F, [#pending_ack{tag = CTag,
                                    queue = QName,
                                    msg_id = MsgId}], Acc) ->
+    %% TODO: fix this abstraction leak
     %% quorum queue, needs the consumer tag
     F({QName, CTag}, [MsgId], Acc);
 foreach_per_queue(F, UAL, Acc) ->
@@ -2032,6 +2029,7 @@ notify_limiter(Limiter, Acked) ->
      case rabbit_limiter:is_active(Limiter) of
         false -> ok;
         true  -> case lists:foldl(fun (#pending_ack{tag = CTag}, Acc) when is_integer(CTag) ->
+                                          %% TODO: fix absctraction leak
                                           %% Quorum queues use integer CTags
                                           %% classic queues use binaries
                                           %% Quorum queues do not interact
@@ -2792,3 +2790,12 @@ maybe_decrease_global_publishers(#ch{publishing_mode = true}) ->
 
 is_global_qos_permitted() ->
     rabbit_deprecated_features:is_permitted(global_qos).
+
+-spec unsupported_single_active_consumer_error(amqqueue:amqqueue()) -> no_return().
+unsupported_single_active_consumer_error(Q) ->
+    rabbit_misc:protocol_error(
+      resource_locked,
+      "cannot obtain access to locked ~ts. basic.get operations "
+      "are not supported by ~p queues with single active consumer",
+      [rabbit_misc:rs(amqqueue:get_name(Q)),
+       rabbit_queue_type:short_alias_of(amqqueue:get_type(Q))]).

deps/rabbit/src/rabbit_classic_queue.erl

@@ -64,8 +64,32 @@
          send_drained_credit_api_v1/4,
          send_credit_reply/7]).
 
+-export([queue_topology/1,
+         feature_flag_name/0,
+         policy_apply_to_name/0,
+         can_redeliver/0,
+         stop/1,
+         is_replicated/0,
+         rebalance_module/0,
+         list_with_minimum_quorum/0,
+         drain/1,
+         revive/0,
+         queue_vm_stats_sups/0,
+         queue_vm_ets/0,
+         dir_base/0]).
+
 -export([validate_policy/1]).
 
+-rabbit_boot_step(
+   {rabbit_classic_queue_type,
+    [{description, "Classic queue: queue type"},
+     {mfa,      {rabbit_registry, register,
+                    [queue, <<"classic">>, ?MODULE]}},
+     {cleanup,  {rabbit_registry, unregister,
+                 [queue, <<"classic">>]}},
+     {requires, rabbit_registry},
+     {enables,     ?MODULE}]}).
+
 -rabbit_boot_step(
    {?MODULE,
     [{description, "Deprecated queue-master-locator support."
@@ -74,7 +98,7 @@
             [policy_validator, <<"queue-master-locator">>, ?MODULE]}},
      {mfa, {rabbit_registry, register,
             [operator_policy_validator, <<"queue-master-locator">>, ?MODULE]}},
-     {requires, rabbit_registry},
+     {requires, [rabbit_classic_queue_type]},
      {enables, recovery}]}).
 
 validate_policy(Args) ->
@@ -674,3 +698,56 @@ send_credit_reply(Pid, QName, Ctag, DeliveryCount, Credit, Available, Drain) ->
 
 send_queue_event(Pid, QName, Event) ->
     gen_server:cast(Pid, {queue_event, QName, Event}).
+
+-spec queue_topology(amqqueue:amqqueue()) ->
+    {Leader :: undefined | node(), Replicas :: undefined | [node(),...]}.
+queue_topology(Q) ->
+    Pid = amqqueue:get_pid(Q),
+    Node = node(Pid),
+    {Node, [Node]}.
+
+feature_flag_name() ->
+    undefined.
+
+policy_apply_to_name() ->
+    <<"classic_queues">>.
+
+can_redeliver() ->
+    true.
+
+stop(VHost) ->
+    ok = rabbit_amqqueue_sup_sup:stop_for_vhost(VHost),
+    {ok, BQ} = application:get_env(rabbit, backing_queue_module),
+    ok = BQ:stop(VHost).
+
+is_replicated() ->
+    false.
+
+rebalance_module() ->
+    {error, not_supported}.
+
+list_with_minimum_quorum() ->
+    [].
+
+drain(_TransferCandidates) ->
+    ok.
+
+revive() ->
+    ok.
+
+queue_vm_stats_sups() ->
+    {[queue_procs], [rabbit_vm:all_vhosts_children(rabbit_amqqueue_sup_sup)]}.
+
+%% return nothing because of this line in rabbit_vm:
+%% {msg_index,            MsgIndexETS + MsgIndexProc},
+%% it mixes procs and ets,
+%% TODO: maybe instead of separating sups and ets
+%% I need vm_memory callback that just
+%% returns proplist? And rabbit_vm calculates
+%% Other as usual by substraction.
+queue_vm_ets() ->
+    {[],
+     []}.
+
+dir_base() ->
+    [rabbit_vhost:msg_store_dir_base()].

deps/rabbit/src/rabbit_core_metrics_gc.erl

@@ -74,22 +74,15 @@ gc_local_queues() ->
     GbSetDown = gb_sets:from_list(QueuesDown),
     gc_queue_metrics(GbSet, GbSetDown),
     gc_entity(queue_coarse_metrics, GbSet),
-    Followers = gb_sets:from_list([amqqueue:get_name(Q) || Q <- rabbit_amqqueue:list_local_followers() ]),
-    gc_leader_data(Followers).
-
-gc_leader_data(Followers) ->
-    ets:foldl(fun({Id, _, _, _, _}, none) ->
-                      gc_leader_data(Id, queue_coarse_metrics, Followers)
-              end, none, queue_coarse_metrics).
-
-gc_leader_data(Id, Table, GbSet) ->
-    case gb_sets:is_member(Id, GbSet) of
-        true ->
-            ets:delete(Table, Id),
-            none;
-        false ->
-            none
-    end.
+    %% remove coarse metrics for quorum queues without local leader
+    gc_leader_data().
+
+gc_leader_data() ->
+    _ = [begin
+             QName = amqqueue:get_name(Q),
+             rabbit_core_metrics:delete_queue_coarse_metrics(QName)
+         end || Q <- rabbit_amqqueue:list_local_followers()],
+    ok.
 
 gc_global_queues() ->
     GbSet = gb_sets:from_list(rabbit_amqqueue:list_names()),

deps/rabbit/src/rabbit_definitions.erl

@@ -1045,16 +1045,11 @@ list_queues() ->
 
 queue_definition(Q) ->
     #resource{virtual_host = VHost, name = Name} = amqqueue:get_name(Q),
-    Type = case amqqueue:get_type(Q) of
-               rabbit_classic_queue -> classic;
-               rabbit_quorum_queue -> quorum;
-               rabbit_stream_queue -> stream;
-               T -> T
-           end,
+    TypeModule =  amqqueue:get_type(Q),
     #{
         <<"vhost">> => VHost,
         <<"name">> => Name,
-        <<"type">> => Type,
+        <<"type">> => rabbit_registry:lookup_type_name(queue, TypeModule),
         <<"durable">> => amqqueue:is_durable(Q),
         <<"auto_delete">> => amqqueue:is_auto_delete(Q),
         <<"arguments">> => rabbit_misc:amqp_table(amqqueue:get_arguments(Q))

deps/rabbit/src/rabbit_fifo_dlx_worker.erl

@@ -538,14 +538,8 @@ redeliver0(#pending{delivery = Msg0,
 clients_redeliver(Qs, QTypeState) ->
     lists:filter(fun(Q) ->
                          case rabbit_queue_type:module(Q, QTypeState) of
-                             {ok, rabbit_quorum_queue} ->
-                                 % If #enqueue{} Raft command does not get applied
-                                 % rabbit_fifo_client will resend.
-                                 true;
-                             {ok, rabbit_stream_queue} ->
-                                 true;
-                             _ ->
-                                 false
+                             {ok, TypeModule} -> TypeModule:can_redeliver();
+                             _ -> false
                          end
                  end, Qs).