From 48efe552d1bf74ca6134cfb168986afb7edc4b2b Mon Sep 17 00:00:00 2001
From: Andrew Mayorov <encube.ul@gmail.com>
Date: Thu, 9 Feb 2023 14:31:42 +0300
Subject: [PATCH] test(mqtt-bridge): manage test flaps by waiting for flush
 events

---
 .../test/emqx_bridge_mqtt_SUITE.erl           | 44 ++++++++++++-------
 1 file changed, 28 insertions(+), 16 deletions(-)

diff --git a/apps/emqx_bridge/test/emqx_bridge_mqtt_SUITE.erl b/apps/emqx_bridge/test/emqx_bridge_mqtt_SUITE.erl
index c4afa4db2..49b333216 100644
--- a/apps/emqx_bridge/test/emqx_bridge_mqtt_SUITE.erl
+++ b/apps/emqx_bridge/test/emqx_bridge_mqtt_SUITE.erl
@@ -317,10 +317,13 @@ t_mqtt_conn_bridge_egress(_) ->
     RemoteTopic = <<?EGRESS_REMOTE_TOPIC, "/", LocalTopic/binary>>,
     Payload = <<"hello">>,
     emqx:subscribe(RemoteTopic),
-    timer:sleep(100),
-    %% PUBLISH a message to the 'local' broker, as we have only one broker,
-    %% the remote broker is also the local one.
-    emqx:publish(emqx_message:make(LocalTopic, Payload)),
+
+    ?wait_async_action(
+        %% PUBLISH a message to the 'local' broker, as we have only one broker,
+        %% the remote broker is also the local one.
+        emqx:publish(emqx_message:make(LocalTopic, Payload)),
+        #{?snk_kind := buffer_worker_flush_ack}
+    ),
 
     %% we should receive a message on the "remote" broker, with specified topic
     Msg = assert_mqtt_msg_received(RemoteTopic, Payload),
@@ -356,10 +359,13 @@ t_mqtt_conn_bridge_egress_no_payload_template(_) ->
     RemoteTopic = <<?EGRESS_REMOTE_TOPIC, "/", LocalTopic/binary>>,
     Payload = <<"hello">>,
     emqx:subscribe(RemoteTopic),
-    timer:sleep(100),
-    %% PUBLISH a message to the 'local' broker, as we have only one broker,
-    %% the remote broker is also the local one.
-    emqx:publish(emqx_message:make(LocalTopic, Payload)),
+
+    ?wait_async_action(
+        %% PUBLISH a message to the 'local' broker, as we have only one broker,
+        %% the remote broker is also the local one.
+        emqx:publish(emqx_message:make(LocalTopic, Payload)),
+        #{?snk_kind := buffer_worker_flush_ack}
+    ),
 
     %% we should receive a message on the "remote" broker, with specified topic
     Msg = assert_mqtt_msg_received(RemoteTopic),
@@ -444,10 +450,13 @@ t_mqtt_conn_bridge_ingress_and_egress(_) ->
                 }
             ]
     } = request_bridge_metrics(BridgeIDEgress),
-    timer:sleep(100),
-    %% PUBLISH a message to the 'local' broker, as we have only one broker,
-    %% the remote broker is also the local one.
-    emqx:publish(emqx_message:make(LocalTopic, Payload)),
+
+    ?wait_async_action(
+        %% PUBLISH a message to the 'local' broker, as we have only one broker,
+        %% the remote broker is also the local one.
+        emqx:publish(emqx_message:make(LocalTopic, Payload)),
+        #{?snk_kind := buffer_worker_flush_ack}
+    ),
 
     %% we should receive a message on the "remote" broker, with specified topic
     assert_mqtt_msg_received(RemoteTopic, Payload),
@@ -677,10 +686,13 @@ t_mqtt_conn_bridge_egress_reconnect(_) ->
     RemoteTopic = <<?EGRESS_REMOTE_TOPIC, "/", LocalTopic/binary>>,
     Payload0 = <<"hello">>,
     emqx:subscribe(RemoteTopic),
-    timer:sleep(100),
-    %% PUBLISH a message to the 'local' broker, as we have only one broker,
-    %% the remote broker is also the local one.
-    emqx:publish(emqx_message:make(LocalTopic, Payload0)),
+
+    ?wait_async_action(
+        %% PUBLISH a message to the 'local' broker, as we have only one broker,
+        %% the remote broker is also the local one.
+        emqx:publish(emqx_message:make(LocalTopic, Payload0)),
+        #{?snk_kind := buffer_worker_flush_ack}
+    ),
 
     %% we should receive a message on the "remote" broker, with specified topic
     assert_mqtt_msg_received(RemoteTopic, Payload0),