Merge pull request #9006 from rabbitmq/fix-rabbit_stream_queue_SUITE-consume_and_reject-test

rabbit_stream_queue_SUITE: Fix `consume_and_reject` channel close detection

Author: dumbbell

deps/rabbit/test/rabbit_stream_queue_SUITE.erl

@@ -1351,13 +1351,22 @@ consume_and_reject(Config) ->
     subscribe(Ch1, Q, false, 0),
     receive
         {#'basic.deliver'{delivery_tag = DeliveryTag}, _} ->
+            MRef = erlang:monitor(process, Ch1),
             ok = amqp_channel:cast(Ch1, #'basic.reject'{delivery_tag = DeliveryTag,
                                                       requeue      = true}),
-            %% Reject will throw a not implemented exception. As it is a cast operation,
-            %% we'll detect the conneciton/channel closure on the next call.
-            %% Let's try to redeclare and see what happens
-            ?assertExit({{shutdown, {connection_closing, {server_initiated_close, 540, _}}}, _},
-                        declare(Config, Server, Q, [{<<"x-queue-type">>, longstr, <<"stream">>}]))
+            %% Reject will throw a not implemented exception. As it is a cast
+            %% operation, we detect the connection error from the channel
+            %% process exit reason.
+            receive
+                {'DOWN', MRef, _, _, Reason} ->
+                    ?assertMatch(
+                       {shutdown,
+                        {connection_closing,
+                         {server_initiated_close, 540, _}}},
+                       Reason)
+            after 10000 ->
+                      exit(timeout)
+            end
     after 10000 ->
             exit(timeout)
     end,

deps/rabbitmq_ct_client_helpers/src/rabbit_ct_client_helpers.erl

@@ -217,6 +217,8 @@ open_channel(Config, Node) ->
     Pid ! {open_channel, self()},
     receive
         Ch when is_pid(Ch) -> Ch
+    after 60_000 ->
+        ct:fail("Timed out waiting for connection to open")
     end.
 
 open_connection_and_channel(Config) ->