From 5de74a1592b488f847f06b6b2c48f0b975229d0d Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Tue, 21 Jul 2026 14:20:20 +0300 Subject: [PATCH] [#118] Call adapter.close() from Grizzly WebSocket close callbacks The Grizzly server endpoint (OpenICFWebSocketApplication.OpenICFWebSocket) and the client endpoint (ClientRemoteConnectorInfoManager.ICFWebSocket) drained their OperationMessageListeners on close but never called adapter.close(), so the closeListener registered by WebSocketConnectionGroup.handshake() never fired and every reconnect of the same session left a stale holder in the group: unbounded growth of webSockets, dead holders probed by the trySendMessage round-robin, and a half-closed (CLOSING) Grizzly socket still reported operational. Mirror the Jetty endpoint (#115): fire the holder's close listeners from the transport close callback, in a finally so a throwing listener cannot skip the cleanup. Duplicate close deliveries stay harmless: both listener queues drain via poll() and SimpleWebSocket.close() is a no-op once the state has left CONNECTED. - ReconnectGroupTest (server): group-level regression with a stubbed ProtocolHandler; fails without the fix because the group keeps the stale holder and even reports it operational in the CLOSING state. - ClientReconnectGroupTest (client): real client/server pair; closes one connection at the transport level and waits for the group to return to exactly the permitted holder count after reconnect - times out without the fix. --- .../ClientRemoteConnectorInfoManager.java | 10 +- .../grizzly/OpenICFWebSocketApplication.java | 10 +- .../server/ClientReconnectGroupTest.java | 192 ++++++++++++++++++ .../server/grizzly/ReconnectGroupTest.java | 167 +++++++++++++++ 4 files changed, 375 insertions(+), 4 deletions(-) create mode 100644 OpenICF-java-framework/connector-server-grizzly/src/test/java/org/forgerock/openicf/framework/server/ClientReconnectGroupTest.java create mode 100644 OpenICF-java-framework/connector-server-grizzly/src/test/java/org/forgerock/openicf/framework/server/grizzly/ReconnectGroupTest.java 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"); + } +}