Merge pull request wildfly#18280 from kabir/amqp-startup-race

[WFLY-19824] Workaround for AMQP start problem
darranl · Oct 7, 2024 · 716ff1f · 716ff1f
2 parents 73138f1 + ae03a3b
commit 716ff1f
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 2 deletions.
diff --git a/...tegration/microprofile/reactive/messaging/multiple/deployment/amqp/AmqpMessagingBean.java b/...tegration/microprofile/reactive/messaging/multiple/deployment/amqp/AmqpMessagingBean.java
@@ -32,7 +32,20 @@ public void sink(String word) {
 
     public void send(String word) {
         System.out.println("Sending " + word);
-        emitter.send(word);
+        long end = System.currentTimeMillis() + 30000;
+        // Workaround
+        // TODO https://issues.redhat.com/browse/WFLY-19825 Remove this
+        while (!emitter.hasRequests() && System.currentTimeMillis() < end) {
+            try {
+                Thread.sleep(100);
+            } catch (InterruptedException e) {
+            }
+        }
+        if (emitter.hasRequests()) {
+            emitter.send(word);
+        } else {
+            throw new IllegalStateException("Emitter was not ready in 30 seconds");
+        }
     }
 
     public List<String> getReceived() {

diff --git a/...ntegration/microprofile/reactive/messaging/multiple/earmodule/amqp/AmqpMessagingBean.java b/...ntegration/microprofile/reactive/messaging/multiple/earmodule/amqp/AmqpMessagingBean.java
@@ -32,7 +32,20 @@ public void sink(String word) {
 
     public void send(String word) {
         System.out.println("Sending " + word);
-        emitter.send(word);
+        // Workaround
+        // TODO https://issues.redhat.com/browse/WFLY-19825 Remove this
+        long end = System.currentTimeMillis() + 30000;
+        while (!emitter.hasRequests() && System.currentTimeMillis() < end) {
+            try {
+                Thread.sleep(100);
+            } catch (InterruptedException e) {
+            }
+        }
+        if (emitter.hasRequests()) {
+            emitter.send(word);
+        } else {
+            throw new IllegalStateException("Emitter was not ready in 30 seconds");
+        }
     }
 
     public List<String> getReceived() {