Refactor to include retries

2 files changed, 27 insertions, 26 deletions

diff --git a/test/confirms_rejects_SUITE.erl b/test/confirms_rejects_SUITE.erl
index c529c44418..aaaeb4a939 100644
--- a/test/confirms_rejects_SUITE.erl
+++ b/test/confirms_rejects_SUITE.erl
@@ -135,17 +135,9 @@ dead_queue_rejects(Config) ->
         error(timeout_waiting_for_initial_ack)
     end,
 
-    kill_the_queue(QueueName, Config),
-
-    amqp_channel:cast(Ch, #'basic.publish'{routing_key = QueueName},
-                      #amqp_msg{payload = <<"HI">>}),
-
-    receive
-        {'basic.ack',_,_} -> error(expecting_nack_got_ack);
-        {'basic.nack',_,_,_} -> ok
-    after 10000 ->
-        error(timeout_waiting_for_nack)
-    end.
+    BasicPublish = #'basic.publish'{routing_key = QueueName},
+    AmqpMsg = #amqp_msg{payload = <<"HI">>},
+    kill_queue_expect_nack(Config, Ch, QueueName, BasicPublish, AmqpMsg, 5).
 
 mixed_dead_alive_queues_reject(Config) ->
     Conn = ?config(conn, Config),
@@ -184,17 +176,28 @@ mixed_dead_alive_queues_reject(Config) ->
         error({timeout_waiting_for_initial_ack, process_info(self(), messages)})
     end,
 
-    kill_the_queue(QueueNameDead, Config),
+    BasicPublish = #'basic.publish'{exchange = ExchangeName, routing_key = <<"route">>},
+    AmqpMsg = #amqp_msg{payload = <<"HI">>},
+    kill_queue_expect_nack(Config, Ch, QueueNameDead, BasicPublish, AmqpMsg, 5).
 
-    amqp_channel:cast(Ch, #'basic.publish'{exchange = ExchangeName,
-                                           routing_key = <<"route">>},
-                      #amqp_msg{payload = <<"HI">>}),
-
-    receive
-        {'basic.nack',_,_,_} -> ok;
-        {'basic.ack',_,_} -> error(expecting_nack_got_ack)
-    after 10000 ->
-        error({timeout_waiting_for_nack, process_info(self(), messages)})
+kill_queue_expect_nack(_Config, _Ch, _QueueName, _BasicPublish, _AmqpMsg, 0) ->
+    error(expecting_nack_got_ack);
+kill_queue_expect_nack(Config, Ch, QueueName, BasicPublish, AmqpMsg, Tries) ->
+    kill_the_queue(QueueName, Config),
+    amqp_channel:cast(Ch, BasicPublish, AmqpMsg),
+    R = receive
+            {'basic.nack',_,_,_} ->
+                ok;
+            {'basic.ack',_,_} ->
+                retry
+        after 10000 ->
+                  error({timeout_waiting_for_nack, process_info(self(), messages)})
+        end,
+    case R of
+        ok ->
+            ok;
+        retry ->
+            kill_queue_expect_nack(Config, Ch, QueueName, BasicPublish, AmqpMsg, Tries - 1)
     end.
 
 confirms_rejects_conflict(Config) ->
diff --git a/test/dynamic_ha_SUITE.erl b/test/dynamic_ha_SUITE.erl
index 698646d9c3..c20e2aae0b 100644
--- a/test/dynamic_ha_SUITE.erl
+++ b/test/dynamic_ha_SUITE.erl
@@ -225,8 +225,7 @@ change_policy(Config) ->
     ok.
 
 change_cluster(Config) ->
-    [A, B, C, D, E] = rabbit_ct_broker_helpers:get_node_configs(Config,
-      nodename),
+    [A, B, C, D, E] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
     rabbit_ct_broker_helpers:cluster_nodes(Config, [A, B, C]),
     ACh = rabbit_ct_client_helpers:open_channel(Config, A),
 
@@ -234,13 +233,12 @@ change_cluster(Config) ->
     assert_slaves(A, ?QNAME, {A, ''}),
 
     %% Give it policy exactly 4, it should mirror to all 3 nodes
-    rabbit_ct_broker_helpers:set_ha_policy(Config, A, ?POLICY,
-      {<<"exactly">>, 4}),
+    rabbit_ct_broker_helpers:set_ha_policy(Config, A, ?POLICY, {<<"exactly">>, 4}),
     assert_slaves(A, ?QNAME, {A, [B, C]}, [{A, []}, {A, [B]}, {A, [C]}]),
 
     %% Add D and E, D joins in
     rabbit_ct_broker_helpers:cluster_nodes(Config, [A, D, E]),
-    assert_slaves(A, ?QNAME, {A, [B, C, D]}, [{A, [B, C]}]),
+    assert_slaves(A, ?QNAME, {A, [B, C, D]}, , [{A, [B, C]}, {A, [B, C, E]}]),
 
     %% Remove D, E joins in
     rabbit_ct_broker_helpers:stop_node(Config, D),