diff --git a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/ConnectorFramework.java b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/ConnectorFramework.java index 1878cbf27..5186edf35 100644 --- a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/ConnectorFramework.java +++ b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/ConnectorFramework.java @@ -12,6 +12,7 @@ * information: "Portions copyright [year] [name of copyright owner]". * * Copyright 2015-2016 ForgeRock AS. + * Portions Copyrighted 2026 3A Systems, LLC */ package org.forgerock.openicf.framework; @@ -22,6 +23,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Executor; import java.util.concurrent.Executors; import java.util.concurrent.ExecutorService; import java.util.concurrent.ScheduledExecutorService; @@ -307,6 +309,10 @@ public void executeMessage(Runnable processMessage) { messageExecutor.execute(processMessage); } + public Executor getMessageExecutor() { + return messageExecutor; + } + // ------ LocalConnectorFramework Implementation Start ------ private final ConcurrentMap localConnectorInfoManagerCache = diff --git a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/OpenICFServerAdapter.java b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/OpenICFServerAdapter.java index 9c3d95210..4a6924fd7 100644 --- a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/OpenICFServerAdapter.java +++ b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/OpenICFServerAdapter.java @@ -128,7 +128,12 @@ public void onMessage(WebSocketConnectionHolder socket, String data) { public void onMessage(final WebSocketConnectionHolder socket, final byte[] bytes) { logger.ok("{0} onMessage({1}:bytes)", loggerName(), bytes.length); - connectorFramework.executeMessage(new Runnable() { + // Dispatch messages of one socket in arrival order: an operation + // response must not overtake the handshake response on a freshly + // opened connection (the pre-handshake branches below would drop it). + // Operation requests are handed off to the shared pool inside + // processMessage, so operations of one socket still run concurrently. + socket.executeSerially(connectorFramework.getMessageExecutor(), new Runnable() { public void run() { processMessage(socket, bytes); } @@ -137,14 +142,11 @@ public void run() { public void processMessage(final WebSocketConnectionHolder socket, byte[] bytes){ try { - RemoteMessage message = RemoteMessage.parseFrom(bytes); + final RemoteMessage message = RemoteMessage.parseFrom(bytes); if (logger.isOk()) { logger.ok("{0} onMessage({1})", loggerName(), message.toString()); } - if (!isHandshakeMessage(message) && !isErrorResponse(message)) { - awaitHandshake(socket, message.getMessageId()); - } if (message.hasRequest()) { if (message.getRequest().hasHandshakeMessage()) { if (isClient()) { @@ -155,9 +157,18 @@ public void processMessage(final WebSocketConnectionHolder socket, byte[] bytes) } } else if (socket.isHandHooked()) { if (message.getRequest().hasOperationRequest()) { - processOperationRequest(socket, message.getMessageId(), message - .getRequest().getOperationRequest()); - + // Execute on the shared pool: operations may run for a + // long time and several of them multiplex over one + // socket. Keeping them out of the serial dispatch + // queue preserves their concurrency and keeps later + // messages (e.g. CancelOpRequest) timely. + connectorFramework.executeMessage(new Runnable() { + @Override + public void run() { + processOperationRequest(socket, message.getMessageId(), message + .getRequest().getOperationRequest()); + } + }); } else if (message.getRequest().hasCancelOpRequest()) { processCancelOpRequest(socket, message.getMessageId(), message.getRequest() .getCancelOpRequest()); @@ -204,47 +215,6 @@ public void processMessage(final WebSocketConnectionHolder socket, byte[] bytes) } } - private static boolean isHandshakeMessage(final RemoteMessage message) { - return (message.hasRequest() && message.getRequest().hasHandshakeMessage()) - || (message.hasResponse() && message.getResponse().hasHandshakeMessage()); - } - - /** - * Error responses are handled before the isHandHooked() guards below, so - * they never need to wait for the handshake. - */ - private static boolean isErrorResponse(final RemoteMessage message) { - return message.hasResponse() && message.getResponse().hasError(); - } - - /** - * Messages are processed on a thread pool, so a message received on a - * freshly opened connection can overtake the handshake that arrived just - * before it and would be dropped by the pre-handshake branches below. The - * handshake is being processed concurrently and normally lands within - * milliseconds, so briefly wait for it instead of discarding the message. - */ - private static final long HANDSHAKE_WAIT_MS = 500; - - private void awaitHandshake(final WebSocketConnectionHolder socket, long messageId) { - if (socket.isHandHooked()) { - return; - } - final long deadline = System.currentTimeMillis() + HANDSHAKE_WAIT_MS; - while (!socket.isHandHooked() && System.currentTimeMillis() < deadline) { - try { - Thread.sleep(10); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return; - } - } - if (!socket.isHandHooked()) { - logger.warn("{0} handshake still pending after wait, message('{1}') via socket:{2} ", - loggerName(), messageId, socket.hashCode()); - } - } - protected void handleRemoteMessage(final WebSocketConnectionHolder socket, final RemoteMessage message) { logger.warn("{0} received unknown message('{1}') via socket:{2} ", loggerName(), message.getMessageId(), diff --git a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolder.java b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolder.java index 3a6785541..4ea78edd7 100644 --- a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolder.java +++ b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolder.java @@ -29,10 +29,13 @@ import java.io.Closeable; import java.util.Queue; import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.Executor; +import java.util.concurrent.atomic.AtomicBoolean; import org.forgerock.openicf.common.protobuf.RPCMessages.HandshakeMessage; import org.forgerock.openicf.common.rpc.RemoteConnectionHolder; import org.forgerock.openicf.framework.CloseListener; +import org.identityconnectors.common.logging.Log; import com.google.protobuf.MessageLite; @@ -41,9 +44,72 @@ public abstract class WebSocketConnectionHolder Closeable, RemoteConnectionHolder { + private static final Log logger = Log.getLog(WebSocketConnectionHolder.class); + protected final Queue> listeners = new ConcurrentLinkedQueue>(); + private final Queue dispatchQueue = new ConcurrentLinkedQueue(); + private final AtomicBoolean dispatchScheduled = new AtomicBoolean(false); + + /** + * Runs {@code task} on {@code executor} after every task previously + * submitted to this holder has completed, preserving the submission order. + * At most one task of this holder is in flight at any time, so messages of + * one socket are processed in arrival order while different sockets still + * run concurrently on the shared executor. + *

+ * Tasks must not block waiting for a later message of the same socket: + * that message cannot be dispatched until the current task returns. + *

+ * Arrival order is preserved only because the WebSocket container invokes + * the per-connection message callbacks sequentially (both Jetty and + * Grizzly do): this method keeps the order of its own calls, so + * concurrent callers would enqueue in an undefined order. + */ + public void executeSerially(final Executor executor, final Runnable task) { + dispatchQueue.add(task); + scheduleDispatch(executor); + } + + private void scheduleDispatch(final Executor executor) { + if (dispatchScheduled.compareAndSet(false, true)) { + try { + executor.execute(new Runnable() { + @Override + public void run() { + drainDispatchQueue(executor); + } + }); + } catch (RuntimeException e) { + // Typically RejectedExecutionException on shutdown - allow a + // later submission to try again instead of wedging the queue. + dispatchScheduled.set(false); + throw e; + } + } + } + + private void drainDispatchQueue(final Executor executor) { + try { + Runnable task; + while ((task = dispatchQueue.poll()) != null) { + try { + task.run(); + } catch (Throwable t) { + logger.warn(t, "Failed to process a dispatched message"); + } + } + } finally { + dispatchScheduled.set(false); + // A task may have been added after poll() returned null but before + // the flag was cleared; whoever wins the CAS reschedules the drain. + if (!dispatchQueue.isEmpty()) { + scheduleDispatch(executor); + } + } + } + public boolean receiveHandshake(HandshakeMessage message) { if (null == getRemoteConnectionContext()) { handshake(message); diff --git a/OpenICF-java-framework/connector-framework-server/src/test/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolderTest.java b/OpenICF-java-framework/connector-framework-server/src/test/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolderTest.java new file mode 100644 index 000000000..b5c6df4d1 --- /dev/null +++ b/OpenICF-java-framework/connector-framework-server/src/test/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolderTest.java @@ -0,0 +1,226 @@ +/* + * 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.remote.rpc; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import org.forgerock.openicf.common.protobuf.RPCMessages.HandshakeMessage; +import org.testng.Assert; +import org.testng.annotations.AfterClass; +import org.testng.annotations.Test; + +/** + * Tests the per-socket serial dispatch queue of + * {@link WebSocketConnectionHolder#executeSerially}: tasks of one holder run + * in submission order without overlapping, tasks of different holders still + * run concurrently, a failing task does not stall the queue, and a rejected + * dispatch does not wedge it. + */ +public class WebSocketConnectionHolderTest { + + private final ExecutorService pool = Executors.newFixedThreadPool(8); + + @AfterClass + public void shutdownPool() { + pool.shutdownNow(); + } + + private static WebSocketConnectionHolder newHolder() { + return new WebSocketConnectionHolder() { + + protected void handshake(HandshakeMessage message) { + } + + protected void tryClose() { + } + + public boolean isOperational() { + return true; + } + + public RemoteOperationContext getRemoteConnectionContext() { + return null; + } + + public Future sendBytes(byte[] data) { + return CompletableFuture.completedFuture(null); + } + + public Future sendString(String data) { + return CompletableFuture.completedFuture(null); + } + + public void sendPing(byte[] applicationData) throws Exception { + } + + public void sendPong(byte[] applicationData) throws Exception { + } + }; + } + + @Test(timeOut = 30000) + public void testTasksRunInSubmissionOrderWithoutOverlap() throws Exception { + final WebSocketConnectionHolder holder = newHolder(); + final int taskCount = 1000; + final List executed = new ArrayList(taskCount); + final AtomicInteger inFlight = new AtomicInteger(0); + final AtomicInteger maxInFlight = new AtomicInteger(0); + final CountDownLatch done = new CountDownLatch(taskCount); + + for (int i = 0; i < taskCount; i++) { + final int id = i; + holder.executeSerially(pool, new Runnable() { + public void run() { + int current = inFlight.incrementAndGet(); + maxInFlight.accumulateAndGet(current, Math::max); + // executed is not synchronized on purpose: tasks never + // overlap, and each task's writes are published to the + // next one by the release/acquire chain through + // dispatchScheduled (set(false) after a drain, winning + // CAS before the next) - non-overlap alone would not + // guarantee visibility. + executed.add(id); + inFlight.decrementAndGet(); + done.countDown(); + } + }); + } + + Assert.assertTrue(done.await(20, TimeUnit.SECONDS), "tasks did not finish"); + Assert.assertEquals(maxInFlight.get(), 1, "tasks of one socket overlapped"); + Assert.assertEquals(executed.size(), taskCount); + for (int i = 0; i < taskCount; i++) { + Assert.assertEquals(executed.get(i).intValue(), i, "task order broken at " + i); + } + } + + @Test(timeOut = 30000) + public void testBlockedHolderDoesNotBlockOtherHolder() throws Exception { + final WebSocketConnectionHolder blocked = newHolder(); + final WebSocketConnectionHolder free = newHolder(); + final CountDownLatch blockedStarted = new CountDownLatch(1); + final CountDownLatch release = new CountDownLatch(1); + final CountDownLatch freeDone = new CountDownLatch(1); + final CountDownLatch blockedDone = new CountDownLatch(1); + + blocked.executeSerially(pool, new Runnable() { + public void run() { + blockedStarted.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + }); + blocked.executeSerially(pool, new Runnable() { + public void run() { + blockedDone.countDown(); + } + }); + + Assert.assertTrue(blockedStarted.await(10, TimeUnit.SECONDS)); + + free.executeSerially(pool, new Runnable() { + public void run() { + freeDone.countDown(); + } + }); + + // The other socket keeps working while the first one is busy... + Assert.assertTrue(freeDone.await(10, TimeUnit.SECONDS), + "an unrelated socket was blocked by a busy one"); + // ...and the second task of the busy socket is still pending. + Assert.assertEquals(blockedDone.getCount(), 1, + "second task of the busy socket must wait for the first"); + + release.countDown(); + Assert.assertTrue(blockedDone.await(10, TimeUnit.SECONDS)); + } + + @Test(timeOut = 30000) + public void testFailingTaskDoesNotStallTheQueue() throws Exception { + final WebSocketConnectionHolder holder = newHolder(); + final CountDownLatch survived = new CountDownLatch(1); + + holder.executeSerially(pool, new Runnable() { + public void run() { + throw new RuntimeException("boom"); + } + }); + holder.executeSerially(pool, new Runnable() { + public void run() { + survived.countDown(); + } + }); + + Assert.assertTrue(survived.await(10, TimeUnit.SECONDS), + "a failing task stalled the serial queue"); + } + + @Test(timeOut = 30000) + public void testRejectedDispatchDoesNotWedgeTheQueue() throws Exception { + final WebSocketConnectionHolder holder = newHolder(); + final AtomicBoolean rejectNext = new AtomicBoolean(true); + final Executor rejectingOnce = new Executor() { + public void execute(Runnable command) { + if (rejectNext.getAndSet(false)) { + throw new RejectedExecutionException("simulated shutdown"); + } + pool.execute(command); + } + }; + final List executed = new ArrayList(); + final CountDownLatch done = new CountDownLatch(2); + + try { + holder.executeSerially(rejectingOnce, recordingTask(1, executed, done)); + Assert.fail("the rejection must reach the caller"); + } catch (RejectedExecutionException expected) { + // The task stays queued; the dispatchScheduled flag must have + // been reset so a later submission can reschedule the drain. + } + + holder.executeSerially(rejectingOnce, recordingTask(2, executed, done)); + + Assert.assertTrue(done.await(10, TimeUnit.SECONDS), + "a rejected dispatch wedged the serial queue"); + Assert.assertEquals(executed, Arrays.asList(1, 2), + "both tasks must drain in submission order"); + } + + private static Runnable recordingTask(final int id, final List executed, + final CountDownLatch done) { + return new Runnable() { + public void run() { + executed.add(id); + done.countDown(); + } + }; + } +} diff --git a/OpenICF-java-framework/connector-server-jetty/src/main/java/org/forgerock/openicf/framework/server/jetty/SinglePrincipal.java b/OpenICF-java-framework/connector-server-jetty/src/main/java/org/forgerock/openicf/framework/server/jetty/SinglePrincipal.java index 79c968beb..ab66a7079 100644 --- a/OpenICF-java-framework/connector-server-jetty/src/main/java/org/forgerock/openicf/framework/server/jetty/SinglePrincipal.java +++ b/OpenICF-java-framework/connector-server-jetty/src/main/java/org/forgerock/openicf/framework/server/jetty/SinglePrincipal.java @@ -35,6 +35,7 @@ import org.forgerock.openicf.framework.remote.rpc.RemoteOperationContext; import org.forgerock.openicf.framework.remote.rpc.WebSocketConnectionGroup; import org.forgerock.openicf.framework.remote.rpc.WebSocketConnectionHolder; +import org.forgerock.util.Utils; import org.forgerock.util.promise.Promises; import org.identityconnectors.framework.common.exceptions.ConnectorIOException; @@ -44,6 +45,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; public class SinglePrincipal extends ConnectionPrincipal implements WebSocketPingPongListener, WebSocketListener, WebSocketFrameListener { @@ -73,7 +75,7 @@ public RemoteOperationContext handshake( } protected void doClose() { - + sendExecutor.shutdown(); } @@ -149,6 +151,17 @@ public void onWebSocketFrame(Frame frame) { // threads via getRemoteConnectionContext()/isHandHooked(). private volatile RemoteOperationContext context = null; + // Single send thread per principal: frames must leave in submission order + // (the peer drops e.g. an operation response that overtakes the handshake + // response). Jetty's RemoteEndpoint is thread-safe, but concurrent + // blocking sends may reach the wire in any order. This instance is cached + // by OpenICFWebSocketCreator and serves every connection of this + // principal name, so the executor must survive onWebSocketClose; it is + // shut down in doClose(). The thread is a daemon because cached + // principals are not closed on servlet destroy. + private final ExecutorService sendExecutor = Executors.newSingleThreadExecutor( + Utils.newThreadFactory(null, "OpenICF Jetty WebSocket Send %d", true)); + private final WebSocketConnectionHolder adapter = new WebSocketConnectionHolder() { protected void handshake(RPCMessages.HandshakeMessage message) { @@ -163,17 +176,22 @@ public RemoteOperationContext getRemoteConnectionContext() { return context; } - private final ExecutorService executorService = Executors.newFixedThreadPool(10); - public Future sendBytes(byte[] data) { if (isOperational()) { - return executorService.submit(() -> { - try { - getSession().getRemote().sendBytes(ByteBuffer.wrap(data)); - } catch (IOException e) { - throw new RuntimeException(e); - } - }); + try { + return sendExecutor.submit(() -> { + try { + getSession().getRemote().sendBytes(ByteBuffer.wrap(data)); + } catch (IOException e) { + throw new RuntimeException(e); + } + }); + } catch (RejectedExecutionException e) { + // The principal was closed and doClose() shut the + // executor down. + return Promises.newExceptionPromise(new ConnectorIOException( + "Socket is not connected.")); + } } else { return Promises.newExceptionPromise(new ConnectorIOException( "Socket is not connected.")); @@ -182,13 +200,18 @@ public Future sendBytes(byte[] data) { public Future sendString(String data) { if (isOperational()) { - return executorService.submit(() -> { - try { - getSession().getRemote().sendString(data); - } catch (IOException e) { - throw new RuntimeException(e); - } - }); + try { + return sendExecutor.submit(() -> { + try { + getSession().getRemote().sendString(data); + } catch (IOException e) { + throw new RuntimeException(e); + } + }); + } catch (RejectedExecutionException e) { + return Promises.newExceptionPromise(new ConnectorIOException( + "Socket is not connected.")); + } } else { return Promises.newExceptionPromise(new ConnectorIOException( "Socket is not connected.")); diff --git a/OpenICF-java-framework/connector-server-jetty/src/test/java/org/forgerock/openicf/framework/server/jetty/ReconnectSendTest.java b/OpenICF-java-framework/connector-server-jetty/src/test/java/org/forgerock/openicf/framework/server/jetty/ReconnectSendTest.java new file mode 100644 index 000000000..69769286f --- /dev/null +++ b/OpenICF-java-framework/connector-server-jetty/src/test/java/org/forgerock/openicf/framework/server/jetty/ReconnectSendTest.java @@ -0,0 +1,108 @@ +/* + * 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.jetty; + +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.Method; +import java.lang.reflect.Proxy; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import org.eclipse.jetty.websocket.api.RemoteEndpoint; +import org.eclipse.jetty.websocket.api.Session; +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.testng.Assert; +import org.testng.annotations.Test; + +/** + * Does a SinglePrincipal that has seen one connection closed still send on a + * subsequent connection? OpenICFWebSocketCreator caches SinglePrincipal per + * principal name, so the same instance serves every connection of that name. + */ +public class ReconnectSendTest { + + private static final AtomicInteger SENT = new AtomicInteger(); + + private static Session newSession() { + final RemoteEndpoint remote = (RemoteEndpoint) Proxy.newProxyInstance( + ReconnectSendTest.class.getClassLoader(), + new Class[] { RemoteEndpoint.class }, + new InvocationHandler() { + public Object invoke(Object p, Method m, Object[] a) { + if ("sendBytes".equals(m.getName())) { + SENT.incrementAndGet(); + } + return null; + } + }); + return (Session) Proxy.newProxyInstance( + ReconnectSendTest.class.getClassLoader(), + new Class[] { Session.class }, + new InvocationHandler() { + public Object invoke(Object p, Method m, Object[] a) { + if ("isOpen".equals(m.getName())) { + return Boolean.TRUE; + } + if ("getRemote".equals(m.getName())) { + return remote; + } + return null; + } + }); + } + + @Test(timeOut = 30000) + public void testSendWorksAfterReconnectOnCachedPrincipal() throws Exception { + final AtomicReference holder = + new AtomicReference(); + + OperationMessageListener capturing = (OperationMessageListener) Proxy.newProxyInstance( + ReconnectSendTest.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]); + } + return null; + } + }); + + SinglePrincipal principal = new SinglePrincipal("anonymous", capturing, null, + new ConcurrentHashMap()); + + // --- connection #1 --- + principal.onWebSocketConnect(newSession()); + Future first = holder.get().sendBytes(new byte[] { 1, 2, 3 }); + first.get(5, TimeUnit.SECONDS); + Assert.assertEquals(SENT.get(), 1, "first connection should have sent"); + + principal.onWebSocketClose(1000, "client went away"); + + // --- connection #2 on the SAME cached principal instance --- + principal.onWebSocketConnect(newSession()); + Future second = holder.get().sendBytes(new byte[] { 4, 5, 6 }); + second.get(5, TimeUnit.SECONDS); + + Assert.assertEquals(SENT.get(), 2, + "send after reconnect must reach the wire"); + } +}