feat: add MUX method that waits for a connection.

jpos/src/main/java/org/jpos/iso/ISOSource.java

@@ -41,4 +41,23 @@ void send(ISOMsg m)
      * @return true if source is connected and usable
      */
     boolean isConnected();
+
+    /**
+     * If this ISOSource is connected, this returns true right away. Otherwise, it waits the specified timeout for connection.
+     *
+     * @param timeout the time to wait for a connection, in ms
+     * @return If the ISOSource connected during the specified timeout
+     */
+    default boolean isConnected(long timeout) {
+        if (isConnected()) return true;
+        long end = System.nanoTime() + timeout * 1_000_000L;
+        long sleep = Math.min(500, timeout);
+        while (sleep > 0 && !Thread.currentThread().isInterrupted()) { // Honor interruptions.
+            ISOUtil.sleep(sleep);
+            if (isConnected()) return true;
+            sleep = Math.min(500, (end - System.nanoTime())/1_000_000L);
+        }
+        return false;
+    }
+
 }

jpos/src/main/java/org/jpos/q2/iso/MUXPool.java

@@ -22,7 +22,6 @@
 import org.jpos.core.ConfigurationException;
 import org.jpos.iso.*;
 import org.jpos.q2.QBeanSupport;
-import org.jpos.q2.QFactory;
 import org.jpos.space.Space;
 import org.jpos.space.SpaceFactory;
 import org.jpos.util.NameRegistrar;
@@ -31,6 +30,7 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /**
@@ -290,10 +290,33 @@ public boolean isConnected() {
         return false;
     }
 
+    @Override
+    public boolean isConnected(long timeout) {
+        if (isConnected()) return true;
+        if (timeout <= 0) return false;
+
+        CompletableFuture<Boolean> result = new CompletableFuture<>();
+        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
+            for (MUX m : mux) {
+                executor.execute(() -> {
+                    if (isUsable(m, timeout, executor)) result.complete(true);
+                });
+            }
+            try {
+                return result.get(timeout, TimeUnit.MILLISECONDS);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            } catch (ExecutionException | TimeoutException _) {
+            } finally {
+                executor.shutdownNow();
+            }
+        }
+        return false;
+    }
+
     @SuppressWarnings("unchecked")
     private boolean isUsable (MUX mux) {
-        if (!checkEnabled || !(mux instanceof QMUX))
-            return mux.isConnected();
+        if (!checkEnabled || !(mux instanceof QMUX)) return mux.isConnected();
 
         QMUX qmux = (QMUX) mux;
         String enabledKey = qmux.getName() + ".enabled";
@@ -305,6 +328,48 @@ private boolean isUsable (MUX mux) {
         return mux.isConnected() && sp.rdp (enabledKey) != null;
     }
 
+    @SuppressWarnings("unchecked")
+    private boolean isUsable(MUX mux, long timeout, ExecutorService executor) {
+        if (!checkEnabled || !(mux instanceof QMUX qmux)) return mux.isConnected(timeout);
+
+        long start = System.nanoTime();
+        long timeoutNanos = TimeUnit.MILLISECONDS.toNanos(timeout);
+        long remaining = timeout;
+        String enabledKey = qmux.getName() + ".enabled";
+
+        while (remaining > 0 && !Thread.currentThread().isInterrupted()) { //Honor interruption
+            final long wait = remaining;
+            CompletableFuture<Boolean> conn = CompletableFuture.supplyAsync(() -> mux.isConnected(wait), executor);
+            CompletableFuture<Boolean> enab = CompletableFuture.supplyAsync(() -> {
+                Object ready = sp.rd(enabledKey, wait);
+                String[] readyNames = qmux.getReadyIndicatorNames();
+                if (readyNames != null && readyNames.length == 1) {
+                    // check that 'mux.enabled' entry has the same content as 'ready'
+                    if (ready == sp.rdp (readyNames[0])) return true;
+                    // relax for some time and retry
+                    ISOUtil.sleep(Math.clamp(wait, 0L, 100L));
+                    return ready == sp.rdp (readyNames[0]);
+                }
+                return ready != null;
+            }, executor);
+
+            try {
+                // Parallel wait for both signals within the current window
+                CompletableFuture.allOf(conn, enab).get(wait, TimeUnit.MILLISECONDS);
+                if (conn.join() && enab.join() && isUsable(mux)) return true;
+                // if both conditions don't meet at the same time, try another round, provided there is time.
+            } catch (ExecutionException | TimeoutException _) {
+                // Last non-blocking check before failing
+                return isUsable(mux);
+            } catch (InterruptedException _) {
+                Thread.currentThread().interrupt();
+                return isUsable(mux);
+            }
+            remaining = (timeoutNanos - (System.nanoTime() - start)) / 1_000_000L;
+        }
+        return isUsable(mux);
+    }
+
     private String[] toStringArray (String s) {
         return (s != null && s.length() > 0) ? ISOUtil.toStringArray(s) : null;
     }

jpos/src/main/java/org/jpos/q2/iso/QMUX.java

@@ -36,8 +36,7 @@
 import java.io.IOException;
 import java.io.PrintStream;
 import java.util.*;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 
 /**
  * @author Alejandro Revilla
@@ -146,7 +145,7 @@ public void destroyService () {
     public static MUX getMUX (String name)
         throws NameRegistrar.NotFoundException
     {
-        return (MUX) NameRegistrar.get ("mux."+name);
+        return NameRegistrar.get ("mux."+name);
     }
 
     /**
@@ -512,6 +511,42 @@ public boolean isConnected() {
         }
         return running();
     }
+
+    /**
+     * If this QMUX is connected, this returns true right away. Otherwise, it waits the specified
+     * timeout for one of the configured {@code <ready>} indicators to appear in the space.
+     *
+     * <p>If no {@code <ready>} indicators are configured, there is nothing concrete to wait for,
+     * so this degrades to a non-blocking {@link #running()} check, mirroring {@link #isConnected()}.
+     *
+     * @param timeout the time to wait for a connection, in ms
+     * @return If the QMUX connected during the specified timeout
+     */
+    @Override
+    public boolean isConnected(long timeout) {
+        if (isConnected()) return true;
+        if (timeout <= 0) return false;
+        if (ready == null || ready.length == 0) return running();
+
+        CompletableFuture<Boolean> result = new CompletableFuture<>();
+        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
+            for (String r : ready) {
+                executor.execute(() -> {
+                    if (sp.rd(r, timeout) != null)
+                        result.complete(true);
+                });
+            }
+            try {
+                return result.get(timeout, TimeUnit.MILLISECONDS);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            } catch (ExecutionException | TimeoutException _) {
+            } finally {
+                executor.shutdownNow();
+            }
+        }
+        return false;
+    }
     public void dump (PrintStream p, String indent) {
         p.println (indent + getCountersAsString());
         metrics.dump (p, indent);

jpos/src/main/java/org/jpos/transaction/participant/QueryHost.java

@@ -119,14 +119,6 @@ else if (o instanceof Number)
     }
 
     protected boolean isConnected (MUX mux) {
-        if (!checkConnected || mux.isConnected())
-            return true;
-        long timeout = System.currentTimeMillis() + waitTimeout;
-        while (System.currentTimeMillis() < timeout) {
-            if (mux.isConnected())
-                return true;
-            ISOUtil.sleep (500);
-        }
-        return false;
+        return !checkConnected || mux.isConnected(waitTimeout);
     }
 }