Add test case enable_ff

Author: ansd

deps/rabbit/test/direct_reply_to_amqpl_SUITE.erl

@@ -40,6 +40,7 @@ groups() ->
     [
      {cluster_size_1, [shuffle],
       [
+       enable_ff,
        trace,
        failure_ack_mode,
        failure_multiple_consumers,
@@ -93,6 +94,115 @@ init_per_testcase(Testcase, Config) ->
 end_per_testcase(Testcase, Config) ->
     rabbit_ct_helpers:testcase_finished(Config, Testcase).
 
+%% In mixed version mode, this test case tests enabling the feature flag
+%% while a client consumes from the volatile queue.
+%% Delete this test case when feature flag rabbitmq_4.2.0 becomes required.
+enable_ff(Config) ->
+    RequestQueue = <<"request queue">>,
+    RequestPayload = <<"my request">>,
+    CorrelationId = <<"my correlation ID">>,
+    RequesterCh = rabbit_ct_client_helpers:open_channel(Config),
+    ResponderCh = rabbit_ct_client_helpers:open_channel(Config),
+
+    amqp_channel:subscribe(RequesterCh,
+                           #'basic.consume'{queue = ?REPLY_QUEUE,
+                                            no_ack = true},
+                           self()),
+    CTag = receive #'basic.consume_ok'{consumer_tag = CTag0} -> CTag0
+           end,
+
+    #'queue.declare_ok'{} = amqp_channel:call(
+                              RequesterCh,
+                              #'queue.declare'{queue = RequestQueue}),
+    #'confirm.select_ok'{} = amqp_channel:call(RequesterCh, #'confirm.select'{}),
+    amqp_channel:register_confirm_handler(RequesterCh, self()),
+
+    %% Send the request.
+    amqp_channel:cast(
+      RequesterCh,
+      #'basic.publish'{routing_key = RequestQueue},
+      #amqp_msg{props = #'P_basic'{reply_to = ?REPLY_QUEUE,
+                                   correlation_id = CorrelationId},
+                payload = RequestPayload}),
+    receive #'basic.ack'{} -> ok
+    end,
+
+    %% Receive the request.
+    {#'basic.get_ok'{},
+     #amqp_msg{props = #'P_basic'{reply_to = ReplyTo,
+                                  correlation_id = CorrelationId},
+               payload = RequestPayload}
+    } = amqp_channel:call(ResponderCh, #'basic.get'{queue = RequestQueue}),
+
+    ?assertEqual(#'queue.declare_ok'{queue = ReplyTo,
+                                     message_count = 0,
+                                     consumer_count = 1},
+                 amqp_channel:call(ResponderCh,
+                                   #'queue.declare'{queue = ReplyTo})),
+
+    %% Send the first reply.
+    amqp_channel:cast(
+      ResponderCh,
+      #'basic.publish'{routing_key = ReplyTo},
+      #amqp_msg{props = #'P_basic'{correlation_id = CorrelationId},
+                payload = <<"reply 1">>}),
+
+    %% Receive the frst reply.
+    receive {#'basic.deliver'{consumer_tag = CTag,
+                              redelivered = false,
+                              exchange = <<>>,
+                              routing_key = ReplyTo},
+             #amqp_msg{payload = P1,
+                       props = #'P_basic'{correlation_id = CorrelationId}}} ->
+                ?assertEqual(<<"reply 1">>, P1)
+    after ?TIMEOUT -> ct:fail({missing_reply, ?LINE})
+    end,
+
+    ok = rabbit_ct_broker_helpers:enable_feature_flag(Config, 'rabbitmq_4.2.0'),
+
+    ?assertEqual(#'queue.declare_ok'{queue = ReplyTo,
+                                     message_count = 0,
+                                     consumer_count = 1},
+                 amqp_channel:call(ResponderCh,
+                                   #'queue.declare'{queue = ReplyTo})),
+
+    %% Send the second reply.
+    amqp_channel:cast(
+      ResponderCh,
+      #'basic.publish'{routing_key = ReplyTo},
+      #amqp_msg{props = #'P_basic'{correlation_id = CorrelationId},
+                payload = <<"reply 2">>}),
+
+    %% Receive the second reply.
+    receive {#'basic.deliver'{consumer_tag = CTag},
+             #amqp_msg{payload = P2,
+                       props = #'P_basic'{correlation_id = CorrelationId}}} ->
+                ?assertEqual(<<"reply 2">>, P2)
+    after ?TIMEOUT -> ct:fail({missing_reply, ?LINE})
+    end,
+
+    %% Requester cancels consumption.
+    ?assertMatch(#'basic.cancel_ok'{consumer_tag = CTag},
+                 amqp_channel:call(RequesterCh, #'basic.cancel'{consumer_tag = CTag})),
+
+    %% Responder checks again if the requester is still there.
+    %% This time, the requester and its queue should be gone.
+    try amqp_channel:call(ResponderCh, #'queue.declare'{queue = ReplyTo}) of
+        _ ->
+            ct:fail("expected queue.declare to fail")
+    catch exit:Reason ->
+              ?assertMatch(
+                 {{_, {_, _, <<"NOT_FOUND - no queue '",
+                               ReplyTo:(byte_size(ReplyTo))/binary,
+                               "' in vhost '/'">>}}, _},
+                 Reason)
+    end,
+
+    %% Clean up.
+    #'queue.delete_ok'{} = amqp_channel:call(RequesterCh,
+                                             #'queue.delete'{queue = RequestQueue}),
+    ok = rabbit_ct_client_helpers:close_channel(RequesterCh).
+
 %% Test case for
 %% https://github.com/rabbitmq/rabbitmq-server/discussions/11662
 trace(Config) ->