From ad8ba3685ae122ad937d623ec3f3d6bd5ea66fda Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9s=20Alcarraz?= Date: Tue, 21 Apr 2026 17:30:27 -0500 Subject: [PATCH 1/7] feat: add MUX method that waits for a connection. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Andrés Alcarraz --- jpos/src/main/java/org/jpos/iso/MUX.java | 18 ++++++++++ .../main/java/org/jpos/q2/iso/MUXPool.java | 26 ++++++++++++++ jpos/src/main/java/org/jpos/q2/iso/QMUX.java | 32 +++++++++++++++-- .../transaction/participant/QueryHost.java | 10 +----- .../java/org/jpos/q2/iso/MUXPoolTest.java | 33 ++++++++++++++++++ .../test/java/org/jpos/q2/iso/QMUXTest.java | 34 +++++++++++++++++++ 6 files changed, 141 insertions(+), 12 deletions(-) diff --git a/jpos/src/main/java/org/jpos/iso/MUX.java b/jpos/src/main/java/org/jpos/iso/MUX.java index b860ff7f96..660130be8a 100644 --- a/jpos/src/main/java/org/jpos/iso/MUX.java +++ b/jpos/src/main/java/org/jpos/iso/MUX.java @@ -42,4 +42,22 @@ public interface MUX extends ISOSource { */ void request(ISOMsg m, long timeout, ISOResponseListener r, Object handBack) throws ISOException; + + /** + * If the mux 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 mux was able to connect 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) { + ISOUtil.sleep(sleep); + if (isConnected()) return true; + sleep = Math.min(500, (end - System.nanoTime())/1_000_000L); + } + return false; + } } diff --git a/jpos/src/main/java/org/jpos/q2/iso/MUXPool.java b/jpos/src/main/java/org/jpos/q2/iso/MUXPool.java index 22cfcb8136..25e3b474c8 100644 --- a/jpos/src/main/java/org/jpos/q2/iso/MUXPool.java +++ b/jpos/src/main/java/org/jpos/q2/iso/MUXPool.java @@ -31,6 +31,7 @@ import java.io.IOException; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; /** @@ -290,6 +291,31 @@ public boolean isConnected() { return false; } + @Override + public boolean isConnected(long timeout) { + if (isConnected()) + return true; + + CompletableFuture result = new CompletableFuture<>(); + try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { + for (MUX m : mux) { + executor.execute(() -> { + if (m.isConnected(timeout)) + 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)) diff --git a/jpos/src/main/java/org/jpos/q2/iso/QMUX.java b/jpos/src/main/java/org/jpos/q2/iso/QMUX.java index e0a6fb4c26..6d678fd46f 100644 --- a/jpos/src/main/java/org/jpos/q2/iso/QMUX.java +++ b/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,33 @@ public boolean isConnected() { } return running(); } + + @Override + public boolean isConnected(long timeout) { + if (isConnected()) + return true; + if (ready == null || ready.length == 0) + return running(); + + CompletableFuture 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); diff --git a/jpos/src/main/java/org/jpos/transaction/participant/QueryHost.java b/jpos/src/main/java/org/jpos/transaction/participant/QueryHost.java index be69c19084..0f70248694 100644 --- a/jpos/src/main/java/org/jpos/transaction/participant/QueryHost.java +++ b/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); } } diff --git a/jpos/src/test/java/org/jpos/q2/iso/MUXPoolTest.java b/jpos/src/test/java/org/jpos/q2/iso/MUXPoolTest.java index 9705e1b13f..0555c6e71b 100644 --- a/jpos/src/test/java/org/jpos/q2/iso/MUXPoolTest.java +++ b/jpos/src/test/java/org/jpos/q2/iso/MUXPoolTest.java @@ -21,9 +21,13 @@ import static org.apache.commons.lang3.JavaVersion.JAVA_14; import static org.apache.commons.lang3.SystemUtils.isJavaVersionAtMost; import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; import org.jdom2.Element; import org.jpos.iso.ISOMsg; +import org.jpos.iso.MUX; import org.junit.jupiter.api.Test; public class MUXPoolTest { @@ -87,4 +91,33 @@ public void testStopService() throws Throwable { mUXPool.stopService(); assertNull(mUXPool.getName(), "mUXPool.getName()"); } + + @Test + public void testIsConnectedWithTimeout() throws Exception { + MUXPool pool = new MUXPool(); + MUX m1 = mock(MUX.class); + MUX m2 = mock(MUX.class); + pool.mux = new MUX[] { m1, m2 }; + + when(m1.isConnected(anyLong())).thenAnswer(invocation -> { + Thread.sleep(200); + return true; + }); + when(m2.isConnected(anyLong())).thenReturn(false); + + assertTrue(pool.isConnected(1000), "Pool should be connected if at least one MUX is connected"); + } + + @Test + public void testIsConnectedWithTimeoutFail() throws Exception { + MUXPool pool = new MUXPool(); + MUX m1 = mock(MUX.class); + MUX m2 = mock(MUX.class); + pool.mux = new MUX[] { m1, m2 }; + + when(m1.isConnected(anyLong())).thenReturn(false); + when(m2.isConnected(anyLong())).thenReturn(false); + + assertFalse(pool.isConnected(500), "Pool should not be connected if all MUXes timeout or return false"); + } } diff --git a/jpos/src/test/java/org/jpos/q2/iso/QMUXTest.java b/jpos/src/test/java/org/jpos/q2/iso/QMUXTest.java index a1e673111b..9eed50bebe 100644 --- a/jpos/src/test/java/org/jpos/q2/iso/QMUXTest.java +++ b/jpos/src/test/java/org/jpos/q2/iso/QMUXTest.java @@ -363,4 +363,38 @@ public void testStopServiceThrowsNullPointerException() throws Throwable { assertNull(qMUX.sp, "qMUX.sp"); } } + + @Test + public void testIsConnectedWithTimeout() throws Exception { + QMUX qmux = new QMUX(); + qmux.setName("test-qmux"); + qmux.setServer(new Q2()); + qmux.setConfiguration(new SimpleConfiguration()); + Element persist = new Element("qmux"); + persist.addContent(new Element("space").setText("tspace:testspace")); + persist.addContent(new Element("in").setText("test.in")); + persist.addContent(new Element("out").setText("test.out")); + persist.addContent(new Element("ready").setText("ready1 ready2")); + qmux.setPersist(persist); + qmux.initService(); + qmux.startService(); + try { + // Put a ready indicator in the space in a separate thread to test the wait + Thread.ofVirtual().start(() -> { + try { + Thread.sleep(200); + qmux.getSpace().out("ready2", "true"); + } catch (InterruptedException e) { + // ignore + } + }); + + assertFalse(qmux.isConnected(100), "Should not be connected yet"); + + assertTrue(qmux.isConnected(400), "Should be connected now"); + } finally { + qmux.stopService(); + qmux.destroyService(); + } + } } From f13aa2c48f54f6a142a926465521b9f455089c8a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9s=20Alcarraz?= Date: Tue, 21 Apr 2026 18:00:33 -0500 Subject: [PATCH 2/7] test: Mux.isConnected with negative and zero timeouts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Andrés Alcarraz --- .../java/org/jpos/q2/iso/MUXPoolTest.java | 18 +++++ .../test/java/org/jpos/q2/iso/QMUXTest.java | 71 ++++++++++++++----- 2 files changed, 71 insertions(+), 18 deletions(-) diff --git a/jpos/src/test/java/org/jpos/q2/iso/MUXPoolTest.java b/jpos/src/test/java/org/jpos/q2/iso/MUXPoolTest.java index 0555c6e71b..ed1e69d759 100644 --- a/jpos/src/test/java/org/jpos/q2/iso/MUXPoolTest.java +++ b/jpos/src/test/java/org/jpos/q2/iso/MUXPoolTest.java @@ -120,4 +120,22 @@ public void testIsConnectedWithTimeoutFail() throws Exception { assertFalse(pool.isConnected(500), "Pool should not be connected if all MUXes timeout or return false"); } + + @Test + public void testIsConnectedWithZeroTimeout() throws Exception { + MUXPool pool = new MUXPool(); + MUX m1 = mock(MUX.class); + pool.mux = new MUX[] { m1 }; + when(m1.isConnected(0)).thenReturn(false); + assertFalse(pool.isConnected(0), "Should return false for zero timeout"); + } + + @Test + public void testIsConnectedWithNegativeTimeout() throws Exception { + MUXPool pool = new MUXPool(); + MUX m1 = mock(MUX.class); + pool.mux = new MUX[] { m1 }; + when(m1.isConnected(-1)).thenReturn(false); + assertFalse(pool.isConnected(-1), "Should return false for negative timeout"); + } } diff --git a/jpos/src/test/java/org/jpos/q2/iso/QMUXTest.java b/jpos/src/test/java/org/jpos/q2/iso/QMUXTest.java index 9eed50bebe..c8ad2c38fb 100644 --- a/jpos/src/test/java/org/jpos/q2/iso/QMUXTest.java +++ b/jpos/src/test/java/org/jpos/q2/iso/QMUXTest.java @@ -18,15 +18,6 @@ package org.jpos.q2.iso; -import static org.apache.commons.lang3.JavaVersion.JAVA_14; -import static org.apache.commons.lang3.SystemUtils.isJavaVersionAtMost; -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNull; -import static org.junit.jupiter.api.Assertions.assertThrows; -import static org.junit.jupiter.api.Assertions.assertTrue; -import static org.junit.jupiter.api.Assertions.fail; - import org.jdom2.Element; import org.jpos.core.SimpleConfiguration; import org.jpos.iso.Connector; @@ -37,6 +28,10 @@ import org.jpos.util.Realm; import org.junit.jupiter.api.Test; +import static org.apache.commons.lang3.JavaVersion.JAVA_14; +import static org.apache.commons.lang3.SystemUtils.isJavaVersionAtMost; +import static org.junit.jupiter.api.Assertions.*; + public class QMUXTest { @Test @@ -88,7 +83,7 @@ public void testGetUnhandledQueue() throws Throwable { String result = new QMUX().getUnhandledQueue(); assertNull(result, "result"); } - + @Test public void testInitServiceThrowsNullPointerException() throws Throwable { assertThrows(NullPointerException.class, () -> { @@ -143,7 +138,7 @@ public void testNotifyThrowsNullPointerException() throws Throwable { assertEquals(0, qMUX.listeners.size(), "qMUX.listeners.size()"); } } - + @Test public void testProcessUnhandledThrowsNullPointerException() throws Throwable { QMUX qMUX = new QMUX(); @@ -364,18 +359,23 @@ public void testStopServiceThrowsNullPointerException() throws Throwable { } } + private Element createPersist(String space, String ready) { + Element persist = new Element("qmux"); + persist.addContent(new Element("space").setText(space)); + persist.addContent(new Element("in").setText("test.in")); + persist.addContent(new Element("out").setText("test.out")); + if (ready != null) + persist.addContent(new Element("ready").setText(ready)); + return persist; + } + @Test public void testIsConnectedWithTimeout() throws Exception { QMUX qmux = new QMUX(); qmux.setName("test-qmux"); qmux.setServer(new Q2()); qmux.setConfiguration(new SimpleConfiguration()); - Element persist = new Element("qmux"); - persist.addContent(new Element("space").setText("tspace:testspace")); - persist.addContent(new Element("in").setText("test.in")); - persist.addContent(new Element("out").setText("test.out")); - persist.addContent(new Element("ready").setText("ready1 ready2")); - qmux.setPersist(persist); + qmux.setPersist(createPersist("tspace:testspace", "ready1 ready2")); qmux.initService(); qmux.startService(); try { @@ -391,10 +391,45 @@ public void testIsConnectedWithTimeout() throws Exception { assertFalse(qmux.isConnected(100), "Should not be connected yet"); - assertTrue(qmux.isConnected(400), "Should be connected now"); + assertTrue(qmux.isConnected(1000), "Should be connected now"); + } finally { + qmux.stopService(); + qmux.destroyService(); + } + } + + @Test + public void testIsConnectedWithZeroTimeout() throws Exception { + QMUX qmux = new QMUX(); + qmux.setName("test-zero-timeout"); + qmux.setServer(new Q2()); + qmux.setConfiguration(new SimpleConfiguration()); + qmux.setPersist(createPersist("tspace:testspace-zero", "ready1")); + qmux.initService(); + qmux.startService(); + try { + assertFalse(qmux.isConnected(0), "Should be false for zero timeout if not connected"); + } finally { + qmux.stopService(); + qmux.destroyService(); + } + } + + @Test + public void testIsConnectedWithNegativeTimeout() throws Exception { + QMUX qmux = new QMUX(); + qmux.setName("test-neg-timeout"); + qmux.setServer(new Q2()); + qmux.setConfiguration(new SimpleConfiguration()); + qmux.setPersist(createPersist("tspace:testspace-neg", "ready1")); + qmux.initService(); + qmux.startService(); + try { + assertFalse(qmux.isConnected(-1), "Should be false for negative timeout if not connected"); } finally { qmux.stopService(); qmux.destroyService(); } } } + From b45db21e710f943d9cc9be9708a8a9599fe2677e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9s=20Alcarraz?= Date: Wed, 29 Apr 2026 18:39:32 -0500 Subject: [PATCH 3/7] refactor: Move isConnected(timeout) to ISOSource MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Perhaps the argument that the `ISOSource` wouldn't connect after disconnection is not strong enough to limit the possibility of having it here. For instance, a MUX connected to a client channel adaptor that can receive responses in a different connection. Not a frequent scenario, but a probable one. Signed-off-by: Andrés Alcarraz --- .../src/main/java/org/jpos/iso/ISOSource.java | 19 +++++++++++++++++++ jpos/src/main/java/org/jpos/iso/MUX.java | 18 ------------------ 2 files changed, 19 insertions(+), 18 deletions(-) diff --git a/jpos/src/main/java/org/jpos/iso/ISOSource.java b/jpos/src/main/java/org/jpos/iso/ISOSource.java index b3f1fc03ef..96bacf0f0f 100644 --- a/jpos/src/main/java/org/jpos/iso/ISOSource.java +++ b/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; + } + } diff --git a/jpos/src/main/java/org/jpos/iso/MUX.java b/jpos/src/main/java/org/jpos/iso/MUX.java index 660130be8a..b860ff7f96 100644 --- a/jpos/src/main/java/org/jpos/iso/MUX.java +++ b/jpos/src/main/java/org/jpos/iso/MUX.java @@ -42,22 +42,4 @@ public interface MUX extends ISOSource { */ void request(ISOMsg m, long timeout, ISOResponseListener r, Object handBack) throws ISOException; - - /** - * If the mux 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 mux was able to connect 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) { - ISOUtil.sleep(sleep); - if (isConnected()) return true; - sleep = Math.min(500, (end - System.nanoTime())/1_000_000L); - } - return false; - } } From d32aa40825a6548c198e265b1651c3d85650b4b4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9s=20Alcarraz?= Date: Thu, 30 Apr 2026 09:06:42 -0500 Subject: [PATCH 4/7] perf: return immediately if timeout is 0 or negative. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Andrés Alcarraz --- jpos/src/main/java/org/jpos/q2/iso/MUXPool.java | 9 +++++---- jpos/src/main/java/org/jpos/q2/iso/QMUX.java | 7 +++---- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/jpos/src/main/java/org/jpos/q2/iso/MUXPool.java b/jpos/src/main/java/org/jpos/q2/iso/MUXPool.java index 25e3b474c8..0676eb83e6 100644 --- a/jpos/src/main/java/org/jpos/q2/iso/MUXPool.java +++ b/jpos/src/main/java/org/jpos/q2/iso/MUXPool.java @@ -293,15 +293,16 @@ public boolean isConnected() { @Override public boolean isConnected(long timeout) { - if (isConnected()) - return true; + + if (isConnected()) return true; + if (timeout <= 0) return isConnected(); + CompletableFuture result = new CompletableFuture<>(); try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { for (MUX m : mux) { executor.execute(() -> { - if (m.isConnected(timeout)) - result.complete(true); + if (m.isConnected(timeout)) result.complete(true); }); } try { diff --git a/jpos/src/main/java/org/jpos/q2/iso/QMUX.java b/jpos/src/main/java/org/jpos/q2/iso/QMUX.java index 6d678fd46f..b7582c7043 100644 --- a/jpos/src/main/java/org/jpos/q2/iso/QMUX.java +++ b/jpos/src/main/java/org/jpos/q2/iso/QMUX.java @@ -514,10 +514,9 @@ public boolean isConnected() { @Override public boolean isConnected(long timeout) { - if (isConnected()) - return true; - if (ready == null || ready.length == 0) - return running(); + if (isConnected()) return true; + if (timeout <= 0) return isConnected(); + if (ready == null || ready.length == 0) return running(); CompletableFuture result = new CompletableFuture<>(); try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { From ca34860b6a191198e8490cd60bdf3d200e10363c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9s=20Alcarraz?= Date: Thu, 30 Apr 2026 15:17:25 -0500 Subject: [PATCH 5/7] fix: Consider enabled for isConnected(timeout) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Andrés Alcarraz --- .../main/java/org/jpos/q2/iso/MUXPool.java | 15 ++++- .../java/org/jpos/q2/iso/MUXPoolTest.java | 56 +++++++++++++++++-- 2 files changed, 63 insertions(+), 8 deletions(-) diff --git a/jpos/src/main/java/org/jpos/q2/iso/MUXPool.java b/jpos/src/main/java/org/jpos/q2/iso/MUXPool.java index 0676eb83e6..cb4f80a156 100644 --- a/jpos/src/main/java/org/jpos/q2/iso/MUXPool.java +++ b/jpos/src/main/java/org/jpos/q2/iso/MUXPool.java @@ -293,16 +293,14 @@ public boolean isConnected() { @Override public boolean isConnected(long timeout) { - if (isConnected()) return true; if (timeout <= 0) return isConnected(); - CompletableFuture result = new CompletableFuture<>(); try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { for (MUX m : mux) { executor.execute(() -> { - if (m.isConnected(timeout)) result.complete(true); + if (isUsable(m, timeout)) result.complete(true); }); } try { @@ -332,6 +330,17 @@ private boolean isUsable (MUX mux) { return mux.isConnected() && sp.rdp (enabledKey) != null; } + @SuppressWarnings("unchecked") + private boolean isUsable(MUX mux, long timeout) { + if (checkEnabled && (mux instanceof QMUX qmux)) { + // We assume that the probability of enabled to change from false to true during timeout is small enough + // to not wait until it gets enabled again + if (sp.rdp(qmux.getName() + ".enabled") == null) + return false; + } + return mux.isConnected(timeout) && isUsable(mux); + } + private String[] toStringArray (String s) { return (s != null && s.length() > 0) ? ISOUtil.toStringArray(s) : null; } diff --git a/jpos/src/test/java/org/jpos/q2/iso/MUXPoolTest.java b/jpos/src/test/java/org/jpos/q2/iso/MUXPoolTest.java index ed1e69d759..550b0582b8 100644 --- a/jpos/src/test/java/org/jpos/q2/iso/MUXPoolTest.java +++ b/jpos/src/test/java/org/jpos/q2/iso/MUXPoolTest.java @@ -22,12 +22,14 @@ import static org.apache.commons.lang3.SystemUtils.isJavaVersionAtMost; import static org.junit.jupiter.api.Assertions.*; import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import org.jdom2.Element; import org.jpos.iso.ISOMsg; import org.jpos.iso.MUX; +import org.jpos.space.Space; import org.junit.jupiter.api.Test; public class MUXPoolTest { @@ -103,6 +105,7 @@ public void testIsConnectedWithTimeout() throws Exception { Thread.sleep(200); return true; }); + when(m1.isConnected()).thenReturn(true); when(m2.isConnected(anyLong())).thenReturn(false); assertTrue(pool.isConnected(1000), "Pool should be connected if at least one MUX is connected"); @@ -116,7 +119,9 @@ public void testIsConnectedWithTimeoutFail() throws Exception { pool.mux = new MUX[] { m1, m2 }; when(m1.isConnected(anyLong())).thenReturn(false); + when(m1.isConnected()).thenReturn(false); when(m2.isConnected(anyLong())).thenReturn(false); + when(m2.isConnected()).thenReturn(false); assertFalse(pool.isConnected(500), "Pool should not be connected if all MUXes timeout or return false"); } @@ -127,15 +132,56 @@ public void testIsConnectedWithZeroTimeout() throws Exception { MUX m1 = mock(MUX.class); pool.mux = new MUX[] { m1 }; when(m1.isConnected(0)).thenReturn(false); + when(m1.isConnected()).thenReturn(false); assertFalse(pool.isConnected(0), "Should return false for zero timeout"); } + @SuppressWarnings({"rawtypes", "unchecked"}) @Test - public void testIsConnectedWithNegativeTimeout() throws Exception { + public void testIsConnectedWithTimeoutAndCheckEnabled() throws Exception { MUXPool pool = new MUXPool(); - MUX m1 = mock(MUX.class); - pool.mux = new MUX[] { m1 }; - when(m1.isConnected(-1)).thenReturn(false); - assertFalse(pool.isConnected(-1), "Should return false for negative timeout"); + pool.checkEnabled = true; + Space sp = mock(Space.class); + pool.sp = sp; + + QMUX qmux = mock(QMUX.class); + when(qmux.getName()).thenReturn("qmux1"); + when(qmux.isConnected(anyLong())).thenReturn(true); + when(qmux.isConnected()).thenReturn(true); + when(qmux.getReadyIndicatorNames()).thenReturn(new String[]{"ready1"}); + + pool.mux = new MUX[] { qmux }; + + // Mock Space behavior: qmux1.enabled matches ready1 value (non-blocking) + Object value = new Object(); + when(sp.rdp("qmux1.enabled")).thenReturn(value); + when(sp.rdp("ready1")).thenReturn(value); + + assertTrue(pool.isConnected(1000), "Pool should be connected if QMUX is connected and enabled indicator matches ready indicator"); + } + + @SuppressWarnings({"rawtypes", "unchecked"}) + @Test + public void testIsConnectedWithTimeoutAndCheckEnabledFail() throws Exception { + MUXPool pool = new MUXPool(); + pool.checkEnabled = true; + Space sp = mock(Space.class); + pool.sp = sp; + + QMUX qmux = mock(QMUX.class); + when(qmux.getName()).thenReturn("qmux1"); + when(qmux.isConnected(anyLong())).thenReturn(true); + when(qmux.isConnected()).thenReturn(true); + when(qmux.getReadyIndicatorNames()).thenReturn(new String[]{"ready1"}); + + pool.mux = new MUX[] { qmux }; + + // Mock Space behavior: ready1 is available, but qmux1.enabled does NOT match ready1 value + Object value1 = new Object(); + Object value2 = new Object(); + when(sp.rdp("ready1")).thenReturn(value1); + when(sp.rdp("qmux1.enabled")).thenReturn(value2); + + assertFalse(pool.isConnected(1000), "Pool should NOT be connected if enabled indicator does NOT match ready indicator"); } } From fd7cebb062b5bf32b31a66b50c57b2852e31547b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9s=20Alcarraz?= Date: Thu, 30 Apr 2026 21:35:06 -0500 Subject: [PATCH 6/7] fix: Consider enabled for isConnected(timeout) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Consider the possibility that if the mux is not enabled at first, can be enabled during the timeout window Signed-off-by: Andrés Alcarraz --- .../main/java/org/jpos/q2/iso/MUXPool.java | 53 ++++-- .../java/org/jpos/q2/iso/MUXPoolTest.java | 159 +++++++++++++++++- 2 files changed, 198 insertions(+), 14 deletions(-) diff --git a/jpos/src/main/java/org/jpos/q2/iso/MUXPool.java b/jpos/src/main/java/org/jpos/q2/iso/MUXPool.java index cb4f80a156..7be011964e 100644 --- a/jpos/src/main/java/org/jpos/q2/iso/MUXPool.java +++ b/jpos/src/main/java/org/jpos/q2/iso/MUXPool.java @@ -297,10 +297,10 @@ public boolean isConnected(long timeout) { if (timeout <= 0) return isConnected(); CompletableFuture result = new CompletableFuture<>(); - try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { + try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) { for (MUX m : mux) { executor.execute(() -> { - if (isUsable(m, timeout)) result.complete(true); + if (isUsable(m, timeout, executor)) result.complete(true); }); } try { @@ -317,8 +317,7 @@ public boolean isConnected(long timeout) { @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"; @@ -331,14 +330,46 @@ private boolean isUsable (MUX mux) { } @SuppressWarnings("unchecked") - private boolean isUsable(MUX mux, long timeout) { - if (checkEnabled && (mux instanceof QMUX qmux)) { - // We assume that the probability of enabled to change from false to true during timeout is small enough - // to not wait until it gets enabled again - if (sp.rdp(qmux.getName() + ".enabled") == null) - return false; + private boolean isUsable(MUX mux, long timeout, ExecutorService executor) { + if (!checkEnabled || !(mux instanceof QMUX qmux)) return mux.isConnected(timeout); + + long remaining = timeout; + long end = System.nanoTime() + timeout * 1_000_000L; + String enabledKey = qmux.getName() + ".enabled"; + while (remaining > 0 && ! Thread.currentThread().isInterrupted()) { //Honor interruption + final long wait = remaining; + Future muxConnected = executor.submit(() -> mux.isConnected(wait)); + Future enabled = executor.submit(() -> { + 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; + } else { // relax for some time and retry + ISOUtil.sleep(Math.clamp((end - System.nanoTime()) / 1_000_000L, 0L, 100L)); + return ready == sp.rdp (readyNames[0]); + } + } + return ready != null; + }); + + try { + if (muxConnected.get(remaining, TimeUnit.MILLISECONDS) + && enabled.get(remaining, TimeUnit.MILLISECONDS) + && isUsable(mux) + ) return true; + // if both conditions don't meet at the same time, try another round, provided there is time. + remaining = (end - System.nanoTime()) / 1_000_000L; + } catch (ExecutionException | TimeoutException _) { + // Last non-blocking check before failing + return isUsable(mux); + } catch (InterruptedException _) { + Thread.currentThread().interrupt(); + return isUsable(mux); + } } - return mux.isConnected(timeout) && isUsable(mux); + return false; } private String[] toStringArray (String s) { diff --git a/jpos/src/test/java/org/jpos/q2/iso/MUXPoolTest.java b/jpos/src/test/java/org/jpos/q2/iso/MUXPoolTest.java index 550b0582b8..767d755d90 100644 --- a/jpos/src/test/java/org/jpos/q2/iso/MUXPoolTest.java +++ b/jpos/src/test/java/org/jpos/q2/iso/MUXPoolTest.java @@ -30,6 +30,7 @@ import org.jpos.iso.ISOMsg; import org.jpos.iso.MUX; import org.jpos.space.Space; +import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; public class MUXPoolTest { @@ -94,6 +95,10 @@ public void testStopService() throws Throwable { assertNull(mUXPool.getName(), "mUXPool.getName()"); } + /** + * Test isConnected(timeout) where at least one MUX in the pool connects + * before the timeout expires. + */ @Test public void testIsConnectedWithTimeout() throws Exception { MUXPool pool = new MUXPool(); @@ -111,6 +116,10 @@ public void testIsConnectedWithTimeout() throws Exception { assertTrue(pool.isConnected(1000), "Pool should be connected if at least one MUX is connected"); } + /** + * Test isConnected(timeout) where all MUXes in the pool fail to connect + * within the given timeout. + */ @Test public void testIsConnectedWithTimeoutFail() throws Exception { MUXPool pool = new MUXPool(); @@ -127,15 +136,27 @@ public void testIsConnectedWithTimeoutFail() throws Exception { } @Test - public void testIsConnectedWithZeroTimeout() throws Exception { + public void testIsConnectedWithZeroTimeoutWithDisconnectedUnderlyingMux() throws Exception { MUXPool pool = new MUXPool(); MUX m1 = mock(MUX.class); pool.mux = new MUX[] { m1 }; - when(m1.isConnected(0)).thenReturn(false); when(m1.isConnected()).thenReturn(false); - assertFalse(pool.isConnected(0), "Should return false for zero timeout"); + assertFalse(pool.isConnected(0), "Should return false for zero timeout, because underlying mux is not connected"); } + @Test + public void testIsConnectedWithZeroTimeoutWithConnectedUnderlyingMux() throws Exception { + MUXPool pool = new MUXPool(); + MUX m1 = mock(MUX.class); + pool.mux = new MUX[] { m1 }; + when(m1.isConnected()).thenReturn(true); + assertTrue(pool.isConnected(0), "Should return true for zero timeout, because underlying mux is connected"); + } + + /** + * Test isConnected(timeout) with checkEnabled=true where the QMUX is + * connected, and the enabled indicator in space matches the ready indicator. + */ @SuppressWarnings({"rawtypes", "unchecked"}) @Test public void testIsConnectedWithTimeoutAndCheckEnabled() throws Exception { @@ -160,6 +181,10 @@ public void testIsConnectedWithTimeoutAndCheckEnabled() throws Exception { assertTrue(pool.isConnected(1000), "Pool should be connected if QMUX is connected and enabled indicator matches ready indicator"); } + /** + * Test isConnected(timeout) with checkEnabled=true where the QMUX is + * connected but the enabled indicator in space does NOT match the ready indicator. + */ @SuppressWarnings({"rawtypes", "unchecked"}) @Test public void testIsConnectedWithTimeoutAndCheckEnabledFail() throws Exception { @@ -184,4 +209,132 @@ public void testIsConnectedWithTimeoutAndCheckEnabledFail() throws Exception { assertFalse(pool.isConnected(1000), "Pool should NOT be connected if enabled indicator does NOT match ready indicator"); } + + /** + * Test isConnected(timeout) with checkEnabled=true where the MUX is initially + * connected but not enabled. It waits for both conditions to be met simultaneously + * using the parallel implementation. + */ + @SuppressWarnings({"rawtypes", "unchecked"}) + @Test + public void testIsConnectedWithTimeoutAndParallelCondition() throws Exception { + MUXPool pool = new MUXPool(); + pool.checkEnabled = true; + Space sp = mock(Space.class); + pool.sp = sp; + + QMUX qmux = mock(QMUX.class); + when(qmux.getName()).thenReturn("qmux1"); + when(qmux.getReadyIndicatorNames()).thenReturn(new String[]{"ready1"}); + + pool.mux = new MUX[] { qmux }; + + Object enabled = new Object(); + + // Simulate: initially connected but NOT enabled + when(qmux.isConnected(anyLong())).thenReturn(true); + when(qmux.isConnected()).thenReturn(true); + + // Blocking rd call: simulate it takes some time to become enabled + when(sp.rd(eq("qmux1.enabled"), anyLong())).thenReturn(enabled); + + // isUsable(mux) checks rdp + when(sp.rdp("qmux1.enabled")).thenReturn(null, enabled); // first call null, second call value + when(sp.rdp("ready1")).thenReturn(enabled); + + assertTrue(pool.isConnected(1000), "Pool should eventually be connected when both conditions match"); + } + + /** + * Test isConnected(timeout) with checkEnabled=true where the MUX is enabled + * but never connects. + */ + @SuppressWarnings({"rawtypes", "unchecked"}) + @Test + public void testIsConnectedWithTimeoutEnabledButNotConnected() throws Exception { + MUXPool pool = new MUXPool(); + pool.checkEnabled = true; + Space sp = mock(Space.class); + pool.sp = sp; + + QMUX qmux = mock(QMUX.class); + when(qmux.getName()).thenReturn("qmux1"); + when(qmux.getReadyIndicatorNames()).thenReturn(new String[]{"ready1"}); + pool.mux = new MUX[] { qmux }; + + Object enabled = new Object(); + when(sp.rd("qmux1.enabled", 500L)).thenReturn(enabled); + when(sp.rdp("qmux1.enabled")).thenReturn(enabled); + + // Never connects + when(qmux.isConnected(anyLong())).thenReturn(false); + when(qmux.isConnected()).thenReturn(false); + + assertFalse(pool.isConnected(500), "Should return false if enabled but never connected"); + } + + /** + * Test isConnected(timeout) with checkEnabled=true where the MUX connects + * but is never enabled. + */ + @SuppressWarnings({"rawtypes", "unchecked"}) + @Test + public void testIsConnectedWithTimeoutConnectedButNotEnabled() throws Exception { + MUXPool pool = new MUXPool(); + pool.checkEnabled = true; + Space sp = mock(Space.class); + pool.sp = sp; + + QMUX qmux = mock(QMUX.class); + when(qmux.getName()).thenReturn("qmux1"); + when(qmux.getReadyIndicatorNames()).thenReturn(new String[]{"ready1"}); + pool.mux = new MUX[] { qmux }; + + // Connects + when(qmux.isConnected(anyLong())).thenReturn(true); + when(qmux.isConnected()).thenReturn(true); + when(sp.rdp("ready1")).thenReturn(new Object()); + + // Never enabled + when(sp.rd(eq("qmux1.enabled"), anyLong())).thenReturn(null); + when(sp.rdp("qmux1.enabled")).thenReturn(null); + + assertFalse(pool.isConnected(500), "Should return false if connected but never enabled"); + } + + /** + * Test isConnected(timeout) where it becomes enabled first, then connected. + */ + @SuppressWarnings({"rawtypes", "unchecked"}) + @Test + public void testIsConnectedWithTimeoutEnabledThenConnected() throws Exception { + MUXPool pool = new MUXPool(); + pool.checkEnabled = true; + Space sp = mock(Space.class); + pool.sp = sp; + + QMUX qmux = mock(QMUX.class); + when(qmux.getName()).thenReturn("qmux1"); + when(qmux.getReadyIndicatorNames()).thenReturn(new String[]{"ready1"}); + pool.mux = new MUX[] { qmux }; + + Object val = new Object(); + // Becomes enabled quickly + when(sp.rd(eq("qmux1.enabled"), anyLong())).thenReturn(val); + when(sp.rdp("qmux1.enabled")).thenReturn(val); + + // Becomes connected later + when(qmux.isConnected(anyLong())).thenAnswer(i -> { + try { + Thread.sleep(200); + } catch (InterruptedException e) { + // ignore + } + return true; + }); + when(qmux.isConnected()).thenReturn(false, false, true); + when(sp.rdp("ready1")).thenReturn(val); + + assertTrue(pool.isConnected(1000), "Should return true if it becomes enabled then connected"); + } } From 6f4da3959f4402cc9c9fca4966c89d8b2d323e54 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9s=20Alcarraz?= Date: Wed, 8 Jul 2026 17:56:46 -0500 Subject: [PATCH 7/7] refactor: simplify MUXPool timeout-wait and address PR #708 review feedback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the nested Future-based wait in MUXPool.isUsable(mux, timeout, executor) with CompletableFuture.allOf, so the connected and enabled/ready checks share a single timeout budget instead of the second get() silently getting less time than the first left it. Also documents QMUX.isConnected(timeout)'s fallback to running() when no indicators are configured, with a covering test, and fixes the trailing-whitespace / stray-blank-line issues flagged by review. Signed-off-by: Andrés Alcarraz --- .../src/main/java/org/jpos/iso/ISOSource.java | 2 +- .../main/java/org/jpos/q2/iso/MUXPool.java | 36 +++++++++---------- jpos/src/main/java/org/jpos/q2/iso/QMUX.java | 12 ++++++- .../java/org/jpos/q2/iso/MUXPoolTest.java | 5 ++- .../test/java/org/jpos/q2/iso/QMUXTest.java | 22 +++++++++++- 5 files changed, 52 insertions(+), 25 deletions(-) diff --git a/jpos/src/main/java/org/jpos/iso/ISOSource.java b/jpos/src/main/java/org/jpos/iso/ISOSource.java index 96bacf0f0f..34ec01443c 100644 --- a/jpos/src/main/java/org/jpos/iso/ISOSource.java +++ b/jpos/src/main/java/org/jpos/iso/ISOSource.java @@ -43,7 +43,7 @@ void send(ISOMsg m) boolean isConnected(); /** - * If this ISOSource is connected, this returns true right away. Otherwise, it waits the specified timeout for connection. + * 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 diff --git a/jpos/src/main/java/org/jpos/q2/iso/MUXPool.java b/jpos/src/main/java/org/jpos/q2/iso/MUXPool.java index 7be011964e..34ec2af502 100644 --- a/jpos/src/main/java/org/jpos/q2/iso/MUXPool.java +++ b/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; @@ -294,7 +293,7 @@ public boolean isConnected() { @Override public boolean isConnected(long timeout) { if (isConnected()) return true; - if (timeout <= 0) return isConnected(); + if (timeout <= 0) return false; CompletableFuture result = new CompletableFuture<>(); try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) { @@ -333,34 +332,32 @@ private boolean isUsable (MUX mux) { 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; - long end = System.nanoTime() + timeout * 1_000_000L; String enabledKey = qmux.getName() + ".enabled"; - while (remaining > 0 && ! Thread.currentThread().isInterrupted()) { //Honor interruption + + while (remaining > 0 && !Thread.currentThread().isInterrupted()) { //Honor interruption final long wait = remaining; - Future muxConnected = executor.submit(() -> mux.isConnected(wait)); - Future enabled = executor.submit(() -> { + CompletableFuture conn = CompletableFuture.supplyAsync(() -> mux.isConnected(wait), executor); + CompletableFuture 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; - } else { // relax for some time and retry - ISOUtil.sleep(Math.clamp((end - System.nanoTime()) / 1_000_000L, 0L, 100L)); - return ready == sp.rdp (readyNames[0]); - } + 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 { - if (muxConnected.get(remaining, TimeUnit.MILLISECONDS) - && enabled.get(remaining, TimeUnit.MILLISECONDS) - && isUsable(mux) - ) return true; + // 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. - remaining = (end - System.nanoTime()) / 1_000_000L; } catch (ExecutionException | TimeoutException _) { // Last non-blocking check before failing return isUsable(mux); @@ -368,8 +365,9 @@ && isUsable(mux) Thread.currentThread().interrupt(); return isUsable(mux); } + remaining = (timeoutNanos - (System.nanoTime() - start)) / 1_000_000L; } - return false; + return isUsable(mux); } private String[] toStringArray (String s) { diff --git a/jpos/src/main/java/org/jpos/q2/iso/QMUX.java b/jpos/src/main/java/org/jpos/q2/iso/QMUX.java index b7582c7043..e4bae4663c 100644 --- a/jpos/src/main/java/org/jpos/q2/iso/QMUX.java +++ b/jpos/src/main/java/org/jpos/q2/iso/QMUX.java @@ -512,10 +512,20 @@ 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 } indicators to appear in the space. + * + *

If no {@code } 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 isConnected(); + if (timeout <= 0) return false; if (ready == null || ready.length == 0) return running(); CompletableFuture result = new CompletableFuture<>(); diff --git a/jpos/src/test/java/org/jpos/q2/iso/MUXPoolTest.java b/jpos/src/test/java/org/jpos/q2/iso/MUXPoolTest.java index 767d755d90..b3eb56f053 100644 --- a/jpos/src/test/java/org/jpos/q2/iso/MUXPoolTest.java +++ b/jpos/src/test/java/org/jpos/q2/iso/MUXPoolTest.java @@ -30,7 +30,6 @@ import org.jpos.iso.ISOMsg; import org.jpos.iso.MUX; import org.jpos.space.Space; -import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; public class MUXPoolTest { @@ -154,7 +153,7 @@ public void testIsConnectedWithZeroTimeoutWithConnectedUnderlyingMux() throws Ex } /** - * Test isConnected(timeout) with checkEnabled=true where the QMUX is + * Test isConnected(timeout) with checkEnabled=true where the QMUX is * connected, and the enabled indicator in space matches the ready indicator. */ @SuppressWarnings({"rawtypes", "unchecked"}) @@ -265,7 +264,7 @@ public void testIsConnectedWithTimeoutEnabledButNotConnected() throws Exception Object enabled = new Object(); when(sp.rd("qmux1.enabled", 500L)).thenReturn(enabled); when(sp.rdp("qmux1.enabled")).thenReturn(enabled); - + // Never connects when(qmux.isConnected(anyLong())).thenReturn(false); when(qmux.isConnected()).thenReturn(false); diff --git a/jpos/src/test/java/org/jpos/q2/iso/QMUXTest.java b/jpos/src/test/java/org/jpos/q2/iso/QMUXTest.java index c8ad2c38fb..3f244762cb 100644 --- a/jpos/src/test/java/org/jpos/q2/iso/QMUXTest.java +++ b/jpos/src/test/java/org/jpos/q2/iso/QMUXTest.java @@ -431,5 +431,25 @@ public void testIsConnectedWithNegativeTimeout() throws Exception { qmux.destroyService(); } } -} + @Test + public void testIsConnectedWithTimeoutAndNoReadyIndicators() throws Exception { + QMUX qmux = new QMUX(); + qmux.setName("test-no-ready"); + qmux.setServer(new Q2()); + qmux.setConfiguration(new SimpleConfiguration()); + qmux.setPersist(createPersist("tspace:testspace-no-ready", null)); + qmux.initService(); + qmux.startService(); + try { + long start = System.currentTimeMillis(); + // With no indicators there's nothing to wait for, so this should + // degrade to a non-blocking running() check instead of waiting out the timeout. + assertEquals(qmux.running(), qmux.isConnected(1000)); + assertTrue(System.currentTimeMillis() - start < 500, "Should not block waiting when nothing to wait for"); + } finally { + qmux.stopService(); + qmux.destroyService(); + } + } +}