From fef98551f85f9a5c63d78491925c9e948d7210eb Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Wed, 29 Jul 2026 12:50:15 +0300 Subject: [PATCH 1/2] Notify connection event listeners when an internal connection is closed InternalConnection kept a CopyOnWriteArrayList of ConnectionEventListener that addConnectionEventListener() and removeConnectionEventListener() maintained, but nothing ever iterated: an internal connection never told its listeners anything. isClosed() and isValid() were hardcoded to false and true, each carrying a FIXME saying they should track the close. Replace the list with the existing ConnectionState SPI class, which already implements the VALID -> CLOSED state machine together with the listener notifications, and whose own comment asks for it to be reused by connection implementations. close() now notifies the listeners and forwards to the server connection only on the first call, which also makes it idempotent as Connection.close() requires, and isClosed() and isValid() delegate to the state, resolving both FIXMEs. A listener registered after the connection was closed is notified straight away, matching ConnectionState and GrizzlyLDAPConnection. Connection errors and unsolicited notifications are deliberately not wired up: an internal connection has no transport, so closure by the application is the only event its listeners can observe. Adds InternalConnectionTestCase; four of its five tests fail against the previous code. --- .../opendj/ldap/InternalConnection.java | 31 ++++-- .../ldap/InternalConnectionTestCase.java | 103 ++++++++++++++++++ 2 files changed, 123 insertions(+), 11 deletions(-) create mode 100644 opendj-core/src/test/java/org/forgerock/opendj/ldap/InternalConnectionTestCase.java diff --git a/opendj-core/src/main/java/org/forgerock/opendj/ldap/InternalConnection.java b/opendj-core/src/main/java/org/forgerock/opendj/ldap/InternalConnection.java index f7ec3ec8b1..c808ec254b 100644 --- a/opendj-core/src/main/java/org/forgerock/opendj/ldap/InternalConnection.java +++ b/opendj-core/src/main/java/org/forgerock/opendj/ldap/InternalConnection.java @@ -13,11 +13,10 @@ * * Copyright 2010 Sun Microsystems, Inc. * Portions copyright 2011-2016 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.forgerock.opendj.ldap; -import java.util.List; -import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.atomic.AtomicInteger; import org.forgerock.opendj.ldap.requests.AbandonRequest; @@ -35,6 +34,7 @@ import org.forgerock.opendj.ldap.responses.ExtendedResult; import org.forgerock.opendj.ldap.responses.Result; import org.forgerock.opendj.ldap.spi.BindResultLdapPromiseImpl; +import org.forgerock.opendj.ldap.spi.ConnectionState; import org.forgerock.opendj.ldap.spi.ExtendedResultLdapPromiseImpl; import org.forgerock.opendj.ldap.spi.ResultLdapPromiseImpl; import org.forgerock.opendj.ldap.spi.SearchResultLdapPromiseImpl; @@ -49,7 +49,15 @@ */ final class InternalConnection extends AbstractAsynchronousConnection { private final ServerConnection serverConnection; - private final List listeners = new CopyOnWriteArrayList<>(); + /** + * Tracks whether this connection has been closed and notifies the registered + * connection event listeners when it is. + *

+ * An internal connection has no transport, so it can never fail nor receive + * an unsolicited notification: closure by the application is the only event + * its listeners can observe. + */ + private final ConnectionState state = new ConnectionState(); private final AtomicInteger messageID = new AtomicInteger(); /** @@ -82,7 +90,7 @@ public LdapPromise addAsync(final AddRequest request, @Override public void addConnectionEventListener(final ConnectionEventListener listener) { Reject.ifNull(listener); - listeners.add(listener); + state.addConnectionEventListener(listener); } @Override @@ -97,8 +105,11 @@ public LdapPromise bindAsync(final BindRequest request, @Override public void close(final UnbindRequest request, final String reason) { - final int i = messageID.getAndIncrement(); - serverConnection.handleConnectionClosed(i, request); + // Closing an already closed connection has no effect. + if (state.notifyConnectionClosed()) { + final int i = messageID.getAndIncrement(); + serverConnection.handleConnectionClosed(i, request); + } } @Override @@ -133,14 +144,12 @@ public LdapPromise extendedRequestAsync(final Exte @Override public boolean isClosed() { - // FIXME: this should be true after close has been called. - return false; + return state.isClosed(); } @Override public boolean isValid() { - // FIXME: this should be false if this connection is disconnected. - return true; + return state.isValid(); } @Override @@ -166,7 +175,7 @@ public LdapPromise modifyDNAsync(final ModifyDNRequest request, @Override public void removeConnectionEventListener(final ConnectionEventListener listener) { Reject.ifNull(listener); - listeners.remove(listener); + state.removeConnectionEventListener(listener); } @Override diff --git a/opendj-core/src/test/java/org/forgerock/opendj/ldap/InternalConnectionTestCase.java b/opendj-core/src/test/java/org/forgerock/opendj/ldap/InternalConnectionTestCase.java new file mode 100644 index 0000000000..f0f856ac86 --- /dev/null +++ b/opendj-core/src/test/java/org/forgerock/opendj/ldap/InternalConnectionTestCase.java @@ -0,0 +1,103 @@ +/* + * 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.opendj.ldap; + +import static org.fest.assertions.Assertions.assertThat; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.anyInt; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +import org.forgerock.opendj.ldap.requests.UnbindRequest; +import org.testng.annotations.Test; + +/** Tests the connection life cycle of the pseudo-connection returned by {@link Connections#newInternalConnection}. */ +@SuppressWarnings("javadoc") +public class InternalConnectionTestCase extends SdkTestCase { + + @SuppressWarnings("unchecked") + private static ServerConnection mockServerConnection() { + return mock(ServerConnection.class); + } + + private static void verifyClosedOnce(final ServerConnection serverConnection) { + verify(serverConnection, times(1)).handleConnectionClosed(anyInt(), any(UnbindRequest.class)); + } + + @Test + public void testCloseNotifiesListeners() throws Exception { + final ServerConnection serverConnection = mockServerConnection(); + final Connection connection = Connections.newInternalConnection(serverConnection); + final MockConnectionEventListener listener = new MockConnectionEventListener(); + connection.addConnectionEventListener(listener); + + connection.close(); + + assertThat(listener.getInvocationCount()).isEqualTo(1); + verifyClosedOnce(serverConnection); + } + + @Test + public void testCloseIsIdempotent() throws Exception { + final ServerConnection serverConnection = mockServerConnection(); + final Connection connection = Connections.newInternalConnection(serverConnection); + final MockConnectionEventListener listener = new MockConnectionEventListener(); + connection.addConnectionEventListener(listener); + + connection.close(); + connection.close(); + + assertThat(listener.getInvocationCount()).isEqualTo(1); + verifyClosedOnce(serverConnection); + } + + @Test + public void testRemovedListenerIsNotNotified() throws Exception { + final Connection connection = Connections.newInternalConnection(mockServerConnection()); + final MockConnectionEventListener listener = new MockConnectionEventListener(); + connection.addConnectionEventListener(listener); + connection.removeConnectionEventListener(listener); + + connection.close(); + + assertThat(listener.getInvocationCount()).isEqualTo(0); + } + + /** A listener registered after the connection was closed is told about it straight away. */ + @Test + public void testListenerAddedAfterCloseIsNotifiedImmediately() throws Exception { + final Connection connection = Connections.newInternalConnection(mockServerConnection()); + connection.close(); + + final MockConnectionEventListener listener = new MockConnectionEventListener(); + connection.addConnectionEventListener(listener); + + assertThat(listener.getInvocationCount()).isEqualTo(1); + } + + @Test + public void testConnectionStateReflectsClose() throws Exception { + final Connection connection = Connections.newInternalConnection(mockServerConnection()); + assertThat(connection.isClosed()).isFalse(); + assertThat(connection.isValid()).isTrue(); + + connection.close(); + + assertThat(connection.isClosed()).isTrue(); + assertThat(connection.isValid()).isFalse(); + } +} From 4f1446532f94bb3cab030958ad2072878bd38c17 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Wed, 29 Jul 2026 17:17:47 +0300 Subject: [PATCH 2/2] Address review: safe listener iteration and guaranteed server close ConnectionState iterated its LinkedList of listeners while holding its own monitor, and removeConnectionEventListener() is re-entrant on that monitor. A listener de-registering itself from its own callback - the idiom LDAPConnectionFactory uses - therefore silently skipped the remaining listeners with two listeners registered, and threw ConcurrentModificationException out of the notification loop with three or more. Switch the listener list to CopyOnWriteArrayList, which fixes the close, error and unsolicited notification loops at once, for every connection implementation built on this class. For an internal connection the consequence went beyond a lost notification: the state is set to CLOSED before the listeners are notified, so an exception escaping the loop left ServerConnection.handleConnectionClosed() uncalled while a retried close() was already a no-op - the forward is the only cleanup an internal connection performs, and RequestHandlerFactoryAdapter.doClose() is what cancels the pending requests. Guard the hand-over with an AtomicBoolean and forward it from a finally block, so a misbehaving listener cannot skip it. The exception still reaches the caller, as it does at every other notification site in the SDK. Drop the @throws IllegalStateException from the javadoc of Connection.addConnectionEventListener() and its ConnectionState counterpart: no implementation throws it, they all notify a listener registered after the event immediately, which is what the tests pin down. Adds a test for each failure mode at both levels plus a 16-thread concurrent close, and tightens the existing assertions to check the forwarded unbind request instance rather than any request. All four new tests fail against the previous code. --- .../org/forgerock/opendj/ldap/Connection.java | 7 +- .../opendj/ldap/InternalConnection.java | 22 ++- .../opendj/ldap/spi/ConnectionState.java | 21 ++- .../ldap/InternalConnectionTestCase.java | 135 +++++++++++++++++- .../opendj/ldap/spi/ConnectionStateTest.java | 66 +++++++++ 5 files changed, 234 insertions(+), 17 deletions(-) diff --git a/opendj-core/src/main/java/org/forgerock/opendj/ldap/Connection.java b/opendj-core/src/main/java/org/forgerock/opendj/ldap/Connection.java index aa5ee88323..e7be68c5d8 100644 --- a/opendj-core/src/main/java/org/forgerock/opendj/ldap/Connection.java +++ b/opendj-core/src/main/java/org/forgerock/opendj/ldap/Connection.java @@ -280,13 +280,14 @@ public interface Connection extends Closeable { * Registers the provided connection event listener so that it will be * notified when this connection is closed by the application, receives an * unsolicited notification, or experiences a fatal error. + *

+ * A listener registered once this connection has already failed and/or been + * closed is notified of the events it has missed before this method + * returns. * * @param listener * The listener which wants to be notified when events occur on * this connection. - * @throws IllegalStateException - * If this connection has already been closed, i.e. if - * {@code isClosed() == true}. * @throws NullPointerException * If the {@code listener} was {@code null}. */ diff --git a/opendj-core/src/main/java/org/forgerock/opendj/ldap/InternalConnection.java b/opendj-core/src/main/java/org/forgerock/opendj/ldap/InternalConnection.java index c808ec254b..91d005a966 100644 --- a/opendj-core/src/main/java/org/forgerock/opendj/ldap/InternalConnection.java +++ b/opendj-core/src/main/java/org/forgerock/opendj/ldap/InternalConnection.java @@ -17,6 +17,7 @@ */ package org.forgerock.opendj.ldap; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import org.forgerock.opendj.ldap.requests.AbandonRequest; @@ -58,6 +59,11 @@ final class InternalConnection extends AbstractAsynchronousConnection { * its listeners can observe. */ private final ConnectionState state = new ConnectionState(); + /** + * Guards the hand-over to the server connection so that it happens exactly + * once, even if notifying the listeners fails. + */ + private final AtomicBoolean isClosed = new AtomicBoolean(); private final AtomicInteger messageID = new AtomicInteger(); /** @@ -106,9 +112,19 @@ public LdapPromise bindAsync(final BindRequest request, @Override public void close(final UnbindRequest request, final String reason) { // Closing an already closed connection has no effect. - if (state.notifyConnectionClosed()) { - final int i = messageID.getAndIncrement(); - serverConnection.handleConnectionClosed(i, request); + if (isClosed.compareAndSet(false, true)) { + try { + state.notifyConnectionClosed(); + } finally { + /* + * A listener throwing an exception must not prevent the server + * connection from releasing its resources: for an internal + * connection this is the only cleanup there is, and a second + * close() would be a no-op. + */ + final int i = messageID.getAndIncrement(); + serverConnection.handleConnectionClosed(i, request); + } } } diff --git a/opendj-core/src/main/java/org/forgerock/opendj/ldap/spi/ConnectionState.java b/opendj-core/src/main/java/org/forgerock/opendj/ldap/spi/ConnectionState.java index d24977fab0..3e0c39ac11 100644 --- a/opendj-core/src/main/java/org/forgerock/opendj/ldap/spi/ConnectionState.java +++ b/opendj-core/src/main/java/org/forgerock/opendj/ldap/spi/ConnectionState.java @@ -12,11 +12,12 @@ * information: "Portions Copyright [year] [name of copyright owner]". * * Copyright 2013-2016 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.forgerock.opendj.ldap.spi; -import java.util.LinkedList; import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; import org.forgerock.opendj.ldap.ConnectionEventListener; import org.forgerock.opendj.ldap.LdapException; @@ -245,8 +246,15 @@ void notifyUnsolicitedNotification(final ConnectionState cs, /** Whether the connection has failed due to a disconnect notification. */ private boolean failedDueToDisconnect; - /** Registered event listeners. */ - private final List listeners = new LinkedList<>(); + /** + * Registered event listeners. Copy on write because a listener is allowed + * to register or, more commonly, de-register itself while it is being + * notified: the notification methods are re-entrant on this object's + * monitor, so any other list implementation would have the notification + * loop either skip the remaining listeners or fail with a + * {@link java.util.ConcurrentModificationException}. + */ + private final List listeners = new CopyOnWriteArrayList<>(); /** Internal state implementation. */ private volatile State state = State.VALID; @@ -260,13 +268,14 @@ public ConnectionState() { * Registers the provided connection event listener so that it will be * notified when this connection is closed by the application, receives an * unsolicited notification, or experiences a fatal error. + *

+ * A listener registered once the connection has already failed and/or been + * closed is notified of the events it has missed before this method + * returns. * * @param listener * The listener which wants to be notified when events occur on * this connection. - * @throws IllegalStateException - * If this connection has already been closed, i.e. if - * {@code isClosed() == true}. * @throws NullPointerException * If the {@code listener} was {@code null}. */ diff --git a/opendj-core/src/test/java/org/forgerock/opendj/ldap/InternalConnectionTestCase.java b/opendj-core/src/test/java/org/forgerock/opendj/ldap/InternalConnectionTestCase.java index f0f856ac86..6736af2b51 100644 --- a/opendj-core/src/test/java/org/forgerock/opendj/ldap/InternalConnectionTestCase.java +++ b/opendj-core/src/test/java/org/forgerock/opendj/ldap/InternalConnectionTestCase.java @@ -18,11 +18,27 @@ import static org.fest.assertions.Assertions.assertThat; import static org.mockito.Matchers.any; import static org.mockito.Matchers.anyInt; +import static org.mockito.Matchers.same; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; - +import static org.testng.Assert.fail; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import org.forgerock.opendj.ldap.requests.Requests; import org.forgerock.opendj.ldap.requests.UnbindRequest; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; import org.testng.annotations.Test; /** Tests the connection life cycle of the pseudo-connection returned by {@link Connections#newInternalConnection}. */ @@ -34,6 +50,13 @@ private static ServerConnection mockServerConnection() { return mock(ServerConnection.class); } + /** Asserts that the close was forwarded exactly once, carrying the very request the caller provided. */ + private static void verifyClosedOnce(final ServerConnection serverConnection, + final UnbindRequest request) { + verify(serverConnection, times(1)).handleConnectionClosed(anyInt(), same(request)); + } + + /** Asserts that the close was forwarded exactly once, for callers which let {@code close()} build the request. */ private static void verifyClosedOnce(final ServerConnection serverConnection) { verify(serverConnection, times(1)).handleConnectionClosed(anyInt(), any(UnbindRequest.class)); } @@ -45,10 +68,11 @@ public void testCloseNotifiesListeners() throws Exception { final MockConnectionEventListener listener = new MockConnectionEventListener(); connection.addConnectionEventListener(listener); - connection.close(); + final UnbindRequest request = Requests.newUnbindRequest(); + connection.close(request, null); assertThat(listener.getInvocationCount()).isEqualTo(1); - verifyClosedOnce(serverConnection); + verifyClosedOnce(serverConnection, request); } @Test @@ -58,16 +82,20 @@ public void testCloseIsIdempotent() throws Exception { final MockConnectionEventListener listener = new MockConnectionEventListener(); connection.addConnectionEventListener(listener); - connection.close(); + final UnbindRequest request = Requests.newUnbindRequest(); + connection.close(request, null); + connection.close(request, null); connection.close(); assertThat(listener.getInvocationCount()).isEqualTo(1); + verifyClosedOnce(serverConnection, request); verifyClosedOnce(serverConnection); } @Test public void testRemovedListenerIsNotNotified() throws Exception { - final Connection connection = Connections.newInternalConnection(mockServerConnection()); + final ServerConnection serverConnection = mockServerConnection(); + final Connection connection = Connections.newInternalConnection(serverConnection); final MockConnectionEventListener listener = new MockConnectionEventListener(); connection.addConnectionEventListener(listener); connection.removeConnectionEventListener(listener); @@ -75,6 +103,7 @@ public void testRemovedListenerIsNotNotified() throws Exception { connection.close(); assertThat(listener.getInvocationCount()).isEqualTo(0); + verifyClosedOnce(serverConnection); } /** A listener registered after the connection was closed is told about it straight away. */ @@ -100,4 +129,100 @@ public void testConnectionStateReflectsClose() throws Exception { assertThat(connection.isClosed()).isTrue(); assertThat(connection.isValid()).isFalse(); } + + /** + * De-registering a listener from within its own close notification is a common idiom, and it must neither skip + * the remaining listeners nor break the hand-over to the server connection. + */ + @Test + public void testSelfDeregisteringListenerDoesNotStopNotification() throws Exception { + final ServerConnection serverConnection = mockServerConnection(); + final Connection connection = Connections.newInternalConnection(serverConnection); + final ConnectionEventListener selfDeregistering = mock(ConnectionEventListener.class); + doAnswer(new Answer() { + @Override + public Void answer(final InvocationOnMock invocation) { + connection.removeConnectionEventListener(selfDeregistering); + return null; + } + }).when(selfDeregistering).handleConnectionClosed(); + connection.addConnectionEventListener(selfDeregistering); + + // Three more listeners: with a list which does not tolerate mutation while being iterated, two of them + // would go unnotified and the third would raise a ConcurrentModificationException out of close(). + final List listeners = new ArrayList<>(); + for (int i = 0; i < 3; i++) { + final MockConnectionEventListener listener = new MockConnectionEventListener(); + listeners.add(listener); + connection.addConnectionEventListener(listener); + } + + final UnbindRequest request = Requests.newUnbindRequest(); + connection.close(request, null); + + verify(selfDeregistering, times(1)).handleConnectionClosed(); + for (final MockConnectionEventListener listener : listeners) { + assertThat(listener.getInvocationCount()).isEqualTo(1); + } + verifyClosedOnce(serverConnection, request); + } + + /** + * A misbehaving listener must not prevent the server connection from releasing its resources: the forward is the + * only cleanup an internal connection performs, and a second {@code close()} would be a no-op. + */ + @Test + public void testListenerFailureDoesNotPreventServerClose() throws Exception { + final ServerConnection serverConnection = mockServerConnection(); + final Connection connection = Connections.newInternalConnection(serverConnection); + final ConnectionEventListener listener = mock(ConnectionEventListener.class); + doThrow(new IllegalStateException("listener failure")).when(listener).handleConnectionClosed(); + connection.addConnectionEventListener(listener); + + final UnbindRequest request = Requests.newUnbindRequest(); + try { + connection.close(request, null); + fail("close() should have passed on the listener failure"); + } catch (final IllegalStateException expected) { + // Expected: no notification site in the SDK guards individual listener callbacks. + } + + verifyClosedOnce(serverConnection, request); + assertThat(connection.isClosed()).isTrue(); + } + + /** Concurrent closes notify the listeners and reach the server connection exactly once. */ + @Test + public void testConcurrentCloseNotifiesAndForwardsOnce() throws Exception { + final int threadCount = 16; + final ServerConnection serverConnection = mockServerConnection(); + final Connection connection = Connections.newInternalConnection(serverConnection); + final MockConnectionEventListener listener = new MockConnectionEventListener(); + connection.addConnectionEventListener(listener); + + final CountDownLatch startLatch = new CountDownLatch(1); + final ExecutorService executor = Executors.newFixedThreadPool(threadCount); + try { + final List> futures = new ArrayList<>(threadCount); + for (int i = 0; i < threadCount; i++) { + futures.add(executor.submit(new Callable() { + @Override + public Void call() throws Exception { + startLatch.await(); + connection.close(); + return null; + } + })); + } + startLatch.countDown(); + for (final Future future : futures) { + future.get(30, TimeUnit.SECONDS); + } + } finally { + executor.shutdownNow(); + } + + assertThat(listener.getInvocationCount()).isEqualTo(1); + verifyClosedOnce(serverConnection); + } } diff --git a/opendj-core/src/test/java/org/forgerock/opendj/ldap/spi/ConnectionStateTest.java b/opendj-core/src/test/java/org/forgerock/opendj/ldap/spi/ConnectionStateTest.java index fcff907b09..e4bf2e572c 100644 --- a/opendj-core/src/test/java/org/forgerock/opendj/ldap/spi/ConnectionStateTest.java +++ b/opendj-core/src/test/java/org/forgerock/opendj/ldap/spi/ConnectionStateTest.java @@ -12,6 +12,7 @@ * information: "Portions Copyright [year] [name of copyright owner]". * * Copyright 2013-2016 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.forgerock.opendj.ldap.spi; @@ -242,6 +243,71 @@ public void testValidState() { assertThat(state.getConnectionError()).isNull(); } + /** + * Tests that a listener de-registering itself while being notified of the close, a common idiom, neither skips + * the remaining listeners nor breaks out of the notification loop. + */ + @Test + public void testSelfDeregisteringListenerDuringClose() { + final ConnectionState state = new ConnectionState(); + final ConnectionEventListener selfDeregistering = mock(ConnectionEventListener.class); + doAnswer(new Answer() { + @Override + public Void answer(final InvocationOnMock invocation) { + state.removeConnectionEventListener(selfDeregistering); + return null; + } + }).when(selfDeregistering).handleConnectionClosed(); + + final ConnectionEventListener listener2 = mock(ConnectionEventListener.class); + final ConnectionEventListener listener3 = mock(ConnectionEventListener.class); + final ConnectionEventListener listener4 = mock(ConnectionEventListener.class); + state.addConnectionEventListener(selfDeregistering); + state.addConnectionEventListener(listener2); + state.addConnectionEventListener(listener3); + state.addConnectionEventListener(listener4); + + assertThat(state.notifyConnectionClosed()).isTrue(); + + assertThat(state.isClosed()).isTrue(); + verify(selfDeregistering).handleConnectionClosed(); + verify(listener2).handleConnectionClosed(); + verify(listener3).handleConnectionClosed(); + verify(listener4).handleConnectionClosed(); + } + + /** + * Tests that a listener de-registering itself while being notified of an error does not prevent the remaining + * listeners from being notified. + */ + @Test + public void testSelfDeregisteringListenerDuringError() { + final ConnectionState state = new ConnectionState(); + final ConnectionEventListener selfDeregistering = mock(ConnectionEventListener.class); + doAnswer(new Answer() { + @Override + public Void answer(final InvocationOnMock invocation) { + state.removeConnectionEventListener(selfDeregistering); + return null; + } + }).when(selfDeregistering).handleConnectionError(false, ERROR); + + final ConnectionEventListener listener2 = mock(ConnectionEventListener.class); + final ConnectionEventListener listener3 = mock(ConnectionEventListener.class); + final ConnectionEventListener listener4 = mock(ConnectionEventListener.class); + state.addConnectionEventListener(selfDeregistering); + state.addConnectionEventListener(listener2); + state.addConnectionEventListener(listener3); + state.addConnectionEventListener(listener4); + + assertThat(state.notifyConnectionError(false, ERROR)).isTrue(); + + verify(selfDeregistering).handleConnectionError(false, ERROR); + verify(listener2).handleConnectionError(false, ERROR); + verify(listener3).handleConnectionError(false, ERROR); + verify(listener4).handleConnectionError(false, ERROR); + } + /** * Tests that reentrant close from error listener is handled. */