Improve robustness of the MQTT Dynamic source tests.

smallrye · Aug 24, 2023 · 4235861 · 4235861
1 parent af31781
commit 4235861
Showing 1 changed file with 17 additions and 0 deletions.
diff --git a/...ng-mqtt/src/test/java/io/smallrye/reactive/messaging/mqtt/DynamicMqttTopicSourceTest.java b/...ng-mqtt/src/test/java/io/smallrye/reactive/messaging/mqtt/DynamicMqttTopicSourceTest.java
@@ -48,6 +48,10 @@ private void awaitAndVerify() {
                 .pollInterval(Duration.ofSeconds(1))
                 .until(() -> connector.getReadiness().isOk());
 
+        await()
+                .pollInterval(Duration.ofSeconds(1))
+                .until(() -> connector.getLiveness().isOk());
+
         bean.publish();
 
         await().until(() -> bean.messages().size() >= 3);
@@ -155,23 +159,36 @@ public void publish() {
             emitter.send(MqttMessage
                     .of("/app/hello/mqtt-" + LocalDate.now().toString() + "/greeting", "hello from dynamic topic 1",
                             MqttQoS.EXACTLY_ONCE));
+            nap(10);
             emitter.send(MqttMessage
                     .of("/app/hello/mqtt-" + LocalDate.now().toString() + "/greeting", "hello from dynamic topic 2",
                             MqttQoS.EXACTLY_ONCE));
+            nap(10);
             emitter.send(MqttMessage
                     .of("/app/hello/mqtt-" + LocalDate.now().toString() + "/greeting", "hello from dynamic topic 3",
                             MqttQoS.EXACTLY_ONCE));
+            nap(15);
             emitter.send(MqttMessage
                     .of("$/app/hello/mqtt-" + LocalDate.now().toString() + "/greeting", "hello from dynamic topic 4",
                             MqttQoS.EXACTLY_ONCE));
+            nap(10);
             emitter.send(MqttMessage
                     .of("$/app/hello/mqtt-" + LocalDate.now().toString() + "/greeting", "hello from dynamic topic 5",
                             MqttQoS.EXACTLY_ONCE));
+            nap(5);
             emitter.send(MqttMessage
                     .of("$/app/hello/mqtt-" + LocalDate.now().toString() + "/greeting", "hello from dynamic topic 6",
                             MqttQoS.EXACTLY_ONCE));
         }
 
+        private void nap(int duration) {
+            try {
+                Thread.sleep(duration);
+            } catch (Exception ignored) {
+                // ignored.
+            }
+        }
+
         @Incoming("in")
         public CompletionStage<Void> received(MqttMessage<byte[]> message) {
             messages.add(message);