More fixes to transient queue handling

Author: dumbbell

deps/rabbit/src/rabbit_db_queue.erl

@@ -989,7 +989,7 @@ delete_transient_in_khepri(FilterFun) ->
     case rabbit_khepri:get_many(PathPattern) of
         {ok, Qs} ->
             Items = maps:fold(
-                     fun(Path, #{data := Queue}, Acc) ->
+                     fun(Path, Queue, Acc) when ?is_amqqueue(Queue) ->
                              case FilterFun(Queue) of
                                  true ->
                                      QueueName = khepri_queue_path_to_name(

deps/rabbit/test/bindings_SUITE.erl

@@ -35,10 +35,10 @@ groups() ->
                              from_mnesia_to_khepri
                             ]},
      {mnesia_cluster, [], [
-                           transient_queue_on_node_down_mnesia
+                           transient_queue_on_node_down
                           ]},
      {khepri_cluster, [], [
-                           transient_queue_on_node_down_khepri
+                           transient_queue_on_node_down
                           ]}
     ].
 
@@ -815,7 +815,7 @@ bind_and_delete_exchange_source(Config) ->
                  rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_binding, list, [<<"/">>])),
     ok.
 
-transient_queue_on_node_down_mnesia(Config) ->
+transient_queue_on_node_down(Config) ->
     Server = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
 
     Ch = rabbit_ct_client_helpers:open_channel(Config, Server),
@@ -846,88 +846,33 @@ transient_queue_on_node_down_mnesia(Config) ->
                                    QResource, Q, []),
     DirectAltBinding = binding_record(rabbit_misc:r(<<"/">>, exchange, <<"amq.direct">>),
                                       QAltResource, QAlt, []),
-    Bindings = lists:sort([DefaultBinding, DirectBinding, DefaultAltBinding, DirectAltBinding]),
-
-    ?assertEqual(Bindings,
-                 lists:sort(
-                   rabbit_ct_broker_helpers:rpc(Config, 1, rabbit_binding, list, [<<"/">>]))),
-
-    rabbit_ct_broker_helpers:stop_node(Config, Server),
-
-    Bindings1 = lists:sort([DefaultBinding, DirectBinding]),
-    ?assertEqual([DirectBinding],
-                 lists:sort(rabbit_ct_broker_helpers:rpc(Config, 1, rabbit_binding, list, [<<"/">>]))),
-    ?assertMatch([],
-                 rabbit_ct_broker_helpers:rpc(Config, 1, rabbit_amqqueue, list, [<<"/">>])),
-
-    rabbit_ct_broker_helpers:start_node(Config, Server),
 
+    Bindings1 = lists:sort([DefaultBinding, DirectBinding, DefaultAltBinding, DirectAltBinding]),
     ?awaitMatch(Bindings1,
                 lists:sort(
                   rabbit_ct_broker_helpers:rpc(Config, 1, rabbit_binding, list, [<<"/">>])),
                 30000),
-    ?awaitMatch([_], rabbit_ct_broker_helpers:rpc(Config, 1, rabbit_amqqueue, list, [<<"/">>]),
-                30000),
-    ok.
-
-transient_queue_on_node_down_khepri(Config) ->
-    %% All entities are durable in khepri
-    Server = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
-
-    Ch = rabbit_ct_client_helpers:open_channel(Config, Server),
-    Q = ?config(queue_name, Config),
-    QAlt = ?config(alt_queue_name, Config),
-    ?assertEqual({'queue.declare_ok', Q, 0, 0}, declare(Ch, Q, [])),
-    ?assertEqual({'queue.declare_ok', QAlt, 0, 0}, declare(Ch, QAlt, [], false)),
-
-    DefaultExchange = rabbit_misc:r(<<"/">>, exchange, <<>>),
-    QResource = rabbit_misc:r(<<"/">>, queue, Q),
-    QAltResource = rabbit_misc:r(<<"/">>, queue, QAlt),
-    DefaultBinding = binding_record(DefaultExchange, QResource, Q, []),
-    DefaultAltBinding = binding_record(DefaultExchange, QAltResource, QAlt, []),
-
-    %% Binding to the default exchange, it's always present
-    ?assertEqual(lists:sort([DefaultBinding, DefaultAltBinding]),
-                 lists:sort(rabbit_ct_broker_helpers:rpc(Config, 1, rabbit_binding, list, [<<"/">>]))),
-
-    %% Let's bind to other exchange
-    #'queue.bind_ok'{} = amqp_channel:call(Ch, #'queue.bind'{exchange = <<"amq.direct">>,
-                                                             queue = Q,
-                                                             routing_key = Q}),
-    #'queue.bind_ok'{} = amqp_channel:call(Ch, #'queue.bind'{exchange = <<"amq.direct">>,
-                                                             queue = QAlt,
-                                                             routing_key = QAlt}),
-
-    DirectBinding = binding_record(rabbit_misc:r(<<"/">>, exchange, <<"amq.direct">>),
-                                   QResource, Q, []),
-    DirectAltBinding = binding_record(rabbit_misc:r(<<"/">>, exchange, <<"amq.direct">>),
-                                      QAltResource, QAlt, []),
-    Bindings = lists:sort([DefaultBinding, DirectBinding, DefaultAltBinding, DirectAltBinding]),
-
-    ?assertEqual(Bindings,
-                 lists:sort(
-                   rabbit_ct_broker_helpers:rpc(Config, 1, rabbit_binding, list, [<<"/">>]))),
 
     rabbit_ct_broker_helpers:stop_node(Config, Server),
 
-    Bindings1 = lists:sort([DirectBinding]),
-    ?awaitMatch(Bindings1,
+    ?awaitMatch([DirectBinding],
                 lists:sort(
                   rabbit_ct_broker_helpers:rpc(Config, 1, rabbit_binding, list, [<<"/">>])),
                 30000),
     ?awaitMatch([],
-                 rabbit_ct_broker_helpers:rpc(Config, 1, rabbit_amqqueue, list, [<<"/">>]),
+                rabbit_ct_broker_helpers:rpc(Config, 1, rabbit_amqqueue, list, [<<"/">>]),
                 30000),
 
     rabbit_ct_broker_helpers:start_node(Config, Server),
 
     Bindings2 = lists:sort([DefaultBinding, DirectBinding]),
-    ?awaitMatch([_], rabbit_ct_broker_helpers:rpc(Config, 1, rabbit_amqqueue, list, [<<"/">>]),
-                30000),
     ?awaitMatch(Bindings2,
                 lists:sort(
                   rabbit_ct_broker_helpers:rpc(Config, 1, rabbit_binding, list, [<<"/">>])),
                 30000),
+    ?awaitMatch([_],
+                rabbit_ct_broker_helpers:rpc(Config, 1, rabbit_amqqueue, list, [<<"/">>]),
+                30000),
     ok.
 
 %% Internal