diff --git a/jpos/src/main/java/org/jpos/iso/ISOSource.java b/jpos/src/main/java/org/jpos/iso/ISOSource.java index b3f1fc03ef..34ec01443c 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/q2/iso/MUXPool.java b/jpos/src/main/java/org/jpos/q2/iso/MUXPool.java index 22cfcb8136..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; @@ -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 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 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; + // 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; } 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..e4bae4663c 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,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 } 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 false; + 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..b3eb56f053 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,15 @@ 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.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 { @@ -87,4 +93,247 @@ public void testStopService() throws Throwable { mUXPool.stopService(); 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(); + 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(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"); + } + + /** + * 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(); + MUX m1 = mock(MUX.class); + MUX m2 = mock(MUX.class); + 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"); + } + + @Test + public void testIsConnectedWithZeroTimeoutWithDisconnectedUnderlyingMux() throws Exception { + MUXPool pool = new MUXPool(); + MUX m1 = mock(MUX.class); + pool.mux = new MUX[] { m1 }; + when(m1.isConnected()).thenReturn(false); + 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 { + 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: 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"); + } + + /** + * 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 { + 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"); + } + + /** + * 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"); + } } 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..3f244762cb 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(); @@ -363,4 +358,98 @@ public void testStopServiceThrowsNullPointerException() throws Throwable { assertNull(qMUX.sp, "qMUX.sp"); } } + + 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()); + qmux.setPersist(createPersist("tspace:testspace", "ready1 ready2")); + 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(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(); + } + } + + @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(); + } + } }