diff --git a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/client/ClientRemoteConnectorInfoManager.java b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/client/ClientRemoteConnectorInfoManager.java index 8121ecdc..286ab681 100644 --- a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/client/ClientRemoteConnectorInfoManager.java +++ b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/client/ClientRemoteConnectorInfoManager.java @@ -621,8 +621,14 @@ public void onClose(DataFrame frame) { connectPromise.handleException(new ConnectorException("Connection is closed: #" + closing.getCode() + " - " + closing.getReason())); OperationMessageListener listener; - while ((listener = listeners.poll()) != null) { - listener.onClose(adapter, closing.getCode(), closing.getReason()); + try { + while ((listener = listeners.poll()) != null) { + listener.onClose(adapter, closing.getCode(), closing.getReason()); + } + } finally { + // Fires the holder's close listeners so the + // WebSocketConnectionGroup drops this connection. + adapter.close(); } } diff --git a/OpenICF-java-framework/connector-server-grizzly/src/main/java/org/forgerock/openicf/framework/server/grizzly/OpenICFWebSocketApplication.java b/OpenICF-java-framework/connector-server-grizzly/src/main/java/org/forgerock/openicf/framework/server/grizzly/OpenICFWebSocketApplication.java index dc61bb78..e9aab3d0 100644 --- a/OpenICF-java-framework/connector-server-grizzly/src/main/java/org/forgerock/openicf/framework/server/grizzly/OpenICFWebSocketApplication.java +++ b/OpenICF-java-framework/connector-server-grizzly/src/main/java/org/forgerock/openicf/framework/server/grizzly/OpenICFWebSocketApplication.java @@ -258,8 +258,14 @@ public void onClose(DataFrame frame) { final ClosingFrame closing = (ClosingFrame) frame; OperationMessageListener listener; - while ((listener = listeners.poll()) != null) { - listener.onClose(adapter, closing.getCode(), closing.getReason()); + try { + while ((listener = listeners.poll()) != null) { + listener.onClose(adapter, closing.getCode(), closing.getReason()); + } + } finally { + // Fires the holder's close listeners so the + // WebSocketConnectionGroup drops this connection. + adapter.close(); } } diff --git a/OpenICF-java-framework/connector-server-grizzly/src/test/java/org/forgerock/openicf/framework/server/ClientReconnectGroupTest.java b/OpenICF-java-framework/connector-server-grizzly/src/test/java/org/forgerock/openicf/framework/server/ClientReconnectGroupTest.java new file mode 100644 index 00000000..d4ed8a37 --- /dev/null +++ b/OpenICF-java-framework/connector-server-grizzly/src/test/java/org/forgerock/openicf/framework/server/ClientReconnectGroupTest.java @@ -0,0 +1,192 @@ +/* + * The contents of this file are subject to the terms of the Common Development and + * Distribution License (the License). You may not use this file except in compliance with the + * License. + * + * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the + * specific language governing permission and limitations under the License. + * + * When distributing Covered Software, include this CDDL Header Notice in each file and include + * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL + * Header, with the fields enclosed by brackets [] replaced by your own identifying + * information: "Portions copyright [year] [name of copyright owner]". + * + * Copyright 2026 3A Systems, LLC. + */ +package org.forgerock.openicf.framework.server; + +import static org.forgerock.openicf.framework.server.AsyncRemotePlainConnectorInfoManagerTest.KEY_HASH; +import static org.forgerock.openicf.framework.server.AsyncRemotePlainConnectorInfoManagerTest.buildRemoteWSFrameworkConnectionInfo; +import static org.forgerock.openicf.framework.server.AsyncRemotePlainConnectorInfoManagerTest.findFreePort; + +import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.ConcurrentMap; + +import org.forgerock.openicf.common.rpc.RemoteConnectionGroup; +import org.forgerock.openicf.framework.ConnectorFramework; +import org.forgerock.openicf.framework.ConnectorFrameworkFactory; +import org.forgerock.openicf.framework.client.ClientRemoteConnectorInfoManager; +import org.forgerock.openicf.framework.client.RemoteWSFrameworkConnectionInfo; +import org.forgerock.openicf.framework.remote.ConnectionPrincipal; +import org.forgerock.openicf.framework.remote.ReferenceCountedObject; +import org.forgerock.openicf.framework.remote.rpc.WebSocketConnectionGroup; +import org.forgerock.openicf.framework.remote.rpc.WebSocketConnectionHolder; +import org.forgerock.util.Pair; +import org.glassfish.grizzly.http.server.NetworkListener; +import org.glassfish.grizzly.websockets.WebSocket; +import org.identityconnectors.testconnector.TstConnector; +import org.testng.Assert; +import org.testng.Reporter; +import org.testng.annotations.AfterTest; +import org.testng.annotations.BeforeTest; +import org.testng.annotations.Test; + +/** + * The client-side WebSocketConnectionGroup learns about a departed connection + * only through the holder's close listeners + * (OpenIdentityPlatform/OpenICF#118): before the fix ICFWebSocket.onClose in + * ClientRemoteConnectorInfoManager never called adapter.close(), so every + * reconnect of the same session left a stale duplicate holder in the group. + */ +public class ClientReconnectGroupTest { + + public final int PLAIN_PORT = findFreePort(); + + public final RemoteWSFrameworkConnectionInfo CONNECTION_INFO = + buildRemoteWSFrameworkConnectionInfo(false, PLAIN_PORT, "/reconnect"); + + private ConnectorServer connectorServer = null; + + @BeforeTest + public void startServer() throws Exception { + connectorServer = new ConnectorServer() { + protected String getContextPath() { + return "/reconnect"; + } + }; + connectorServer.setConnectorFrameworkFactory(new ConnectorFrameworkFactory()); + connectorServer.setConnectorBundleURLs(Arrays.asList(TstConnector.class + .getProtectionDomain().getCodeSource().getLocation())); + + connectorServer.init(); + connectorServer.addListener("grizzly", NetworkListener.DEFAULT_NETWORK_HOST, PLAIN_PORT); + connectorServer.setKeyHash(KEY_HASH); + connectorServer.start(); + + Reporter.log("Grizzly Server Started on port " + PLAIN_PORT, true); + } + + @AfterTest + public void stopServer() throws Exception { + connectorServer.stop(); + connectorServer.destroy(); + Reporter.log("Grizzly Server Stopped", true); + } + + @Test(timeOut = 180000) + public void testGroupDropsHolderOnTransportClose() throws Exception { + ReferenceCountedObject.Reference framework = + new ConnectorFrameworkFactory().acquire(); + try { + ClientRemoteConnectorInfoManager manager = + (ClientRemoteConnectorInfoManager) framework.get() + .getRemoteConnectionInfoManagerFactory().connect(CONNECTION_INFO); + + final ConcurrentMap groups = + connectionGroupsOf(manager); + final int permitted = CONNECTION_INFO.getExpectedConnectionCount(); + + for (int cycle = 1; cycle <= 2; cycle++) { + await("all permitted connections must be up (cycle " + cycle + ")", + new Callable() { + public Boolean call() throws Exception { + return allHolders(groups).size() == permitted + && operationalHolders(groups).size() == permitted; + } + }); + + final WebSocketConnectionHolder victim = operationalHolders(groups).get(0); + socketOf(victim).close(); + + // The regression: without adapter.close() in + // ICFWebSocket.onClose the victim stays in the group forever + // and the holder count grows on every reconnect. + await("the closed holder must leave the group after reconnect (cycle " + + cycle + ")", + new Callable() { + public Boolean call() throws Exception { + return !allHolders(groups).contains(victim) + && allHolders(groups).size() == permitted + && operationalHolders(groups).size() == permitted; + } + }); + } + } finally { + framework.release(); + } + } + + private static void await(String message, Callable condition) throws Exception { + long deadline = System.currentTimeMillis() + 60000L; + while (System.currentTimeMillis() < deadline) { + if (Boolean.TRUE.equals(condition.call())) { + return; + } + Thread.sleep(100); + } + Assert.fail(message); + } + + @SuppressWarnings("unchecked") + private static ConcurrentMap connectionGroupsOf( + ClientRemoteConnectorInfoManager manager) throws Exception { + Field field = ConnectionPrincipal.class.getDeclaredField("connectionGroups"); + field.setAccessible(true); + return (ConcurrentMap) field.get(manager); + } + + @SuppressWarnings("unchecked") + private static List allHolders( + ConcurrentMap groups) throws Exception { + Field field = RemoteConnectionGroup.class.getDeclaredField("webSockets"); + field.setAccessible(true); + List result = new ArrayList(); + for (WebSocketConnectionGroup group : groups.values()) { + for (Pair pair : + (List>) field.get(group)) { + result.add(pair.getSecond()); + } + } + return result; + } + + private static List operationalHolders( + ConcurrentMap groups) throws Exception { + List result = new ArrayList(); + for (WebSocketConnectionHolder holder : allHolders(groups)) { + if (holder.isOperational()) { + result.add(holder); + } + } + return result; + } + + /** + * The holder is an anonymous inner class of the private ICFWebSocket; its + * synthetic outer-instance field is the only WebSocket-typed handle a + * test can reach to close the connection at the transport level. + */ + private static WebSocket socketOf(WebSocketConnectionHolder holder) throws Exception { + for (Field field : holder.getClass().getDeclaredFields()) { + if (WebSocket.class.isAssignableFrom(field.getType())) { + field.setAccessible(true); + return (WebSocket) field.get(holder); + } + } + throw new AssertionError("No WebSocket handle on " + holder.getClass()); + } +} diff --git a/OpenICF-java-framework/connector-server-grizzly/src/test/java/org/forgerock/openicf/framework/server/grizzly/ReconnectGroupTest.java b/OpenICF-java-framework/connector-server-grizzly/src/test/java/org/forgerock/openicf/framework/server/grizzly/ReconnectGroupTest.java new file mode 100644 index 00000000..e00c1771 --- /dev/null +++ b/OpenICF-java-framework/connector-server-grizzly/src/test/java/org/forgerock/openicf/framework/server/grizzly/ReconnectGroupTest.java @@ -0,0 +1,167 @@ +/* + * The contents of this file are subject to the terms of the Common Development and + * Distribution License (the License). You may not use this file except in compliance with the + * License. + * + * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the + * specific language governing permission and limitations under the License. + * + * When distributing Covered Software, include this CDDL Header Notice in each file and include + * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL + * Header, with the fields enclosed by brackets [] replaced by your own identifying + * information: "Portions copyright [year] [name of copyright owner]". + * + * Copyright 2026 3A Systems, LLC. + */ +package org.forgerock.openicf.framework.server.grizzly; + +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.Method; +import java.lang.reflect.Proxy; +import java.net.URI; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import org.forgerock.openicf.common.protobuf.RPCMessages; +import org.forgerock.openicf.framework.remote.rpc.OperationMessageListener; +import org.forgerock.openicf.framework.remote.rpc.WebSocketConnectionGroup; +import org.forgerock.openicf.framework.remote.rpc.WebSocketConnectionHolder; +import org.forgerock.openicf.framework.server.grizzly.OpenICFWebSocketApplication.OpenICFWebSocket; +import org.forgerock.openicf.framework.server.grizzly.OpenICFWebSocketApplication.SinglePrincipal; +import org.glassfish.grizzly.Buffer; +import org.glassfish.grizzly.CompletionHandler; +import org.glassfish.grizzly.GrizzlyFuture; +import org.glassfish.grizzly.http.HttpContent; +import org.glassfish.grizzly.websockets.ClosingFrame; +import org.glassfish.grizzly.websockets.DataFrame; +import org.glassfish.grizzly.websockets.HandShake; +import org.glassfish.grizzly.websockets.ProtocolHandler; +import org.testng.Assert; +import org.testng.annotations.Test; + +/** + * The connection group learns about a departed connection only through the + * holder's close listeners (OpenIdentityPlatform/OpenICF#118): before the fix + * {@link OpenICFWebSocket#onClose} never called {@code adapter.close()}, so + * every reconnect of the same session left a stale duplicate holder in + * {@link WebSocketConnectionGroup}. + */ +public class ReconnectGroupTest { + + /** + * A ProtocolHandler without a Connection. SimpleWebSocket.onClose invokes + * close()/doClose() on it, which would try to write the close frame to + * the wire (and throw on the null connection) if not stubbed out. + */ + private static ProtocolHandler stubHandler() { + return new ProtocolHandler(false) { + + @Override + public GrizzlyFuture close(int code, String reason) { + return null; + } + + @Override + public void doClose() { + } + + @Override + public GrizzlyFuture send(DataFrame frame, + CompletionHandler completionHandler) { + return null; + } + + @Override + public GrizzlyFuture send(byte[] data) { + return null; + } + + @Override + public GrizzlyFuture send(String data) { + return null; + } + + @Override + public byte[] frame(DataFrame frame) { + return new byte[0]; + } + + @Override + public HandShake createServerHandShake(HttpContent requestContent) { + return null; + } + + @Override + public HandShake createClientHandShake(URI uri) { + return null; + } + + @Override + public DataFrame parse(Buffer buffer) { + return null; + } + + @Override + protected boolean isControlFrame(byte opcode) { + return false; + } + }; + } + + @Test(timeOut = 30000) + public void testGroupDropsHolderOnClose() { + final AtomicReference holder = + new AtomicReference(); + final AtomicInteger closes = new AtomicInteger(); + + OperationMessageListener capturing = (OperationMessageListener) Proxy.newProxyInstance( + ReconnectGroupTest.class.getClassLoader(), + new Class[] { OperationMessageListener.class }, + new InvocationHandler() { + public Object invoke(Object p, Method m, Object[] a) { + if ("onConnect".equals(m.getName())) { + holder.set((WebSocketConnectionHolder) a[0]); + } + if ("onClose".equals(m.getName())) { + closes.incrementAndGet(); + } + return null; + } + }); + + SinglePrincipal principal = new SinglePrincipal(capturing, + new ConcurrentHashMap()); + WebSocketConnectionGroup group = new WebSocketConnectionGroup("session-1"); + RPCMessages.HandshakeMessage handshake = + RPCMessages.HandshakeMessage.newBuilder().setSessionId("session-1").build(); + + // --- connection #1 joins and leaves the group --- + OpenICFWebSocket first = new OpenICFWebSocket(stubHandler(), null, principal); + first.onConnect(); + group.handshake(principal, holder.get(), handshake); + Assert.assertTrue(group.isOperational(), "handshake must register the holder"); + + first.onClose(new ClosingFrame(1000, "client went away")); + Assert.assertFalse(group.isOperational(), + "the group must drop the holder of a closed connection"); + Assert.assertEquals(closes.get(), 1, "close must reach the listener"); + + // Grizzly delivers onClose once per direction of the close handshake + // - the duplicate must stay harmless. + first.onClose(new ClosingFrame(1000, "duplicate")); + Assert.assertFalse(group.isOperational()); + Assert.assertEquals(closes.get(), 1, "a duplicate close event must stay suppressed"); + + // --- reconnect: no stale holder of connection #1 may keep the group + // alive after connection #2 leaves as well --- + OpenICFWebSocket second = new OpenICFWebSocket(stubHandler(), null, principal); + second.onConnect(); + group.handshake(principal, holder.get(), handshake); + Assert.assertTrue(group.isOperational(), "reconnect must register the new holder"); + + second.onClose(new ClosingFrame(1000, "client went away again")); + Assert.assertFalse(group.isOperational(), + "no stale holder may survive a reconnect cycle"); + } +}