Remove capturing lambda from ConnectionManager. Fixes #252

OpenHFT · May 16, 2023 · 5539d47 · 5539d47
1 parent c655f74
commit 5539d47
Show file tree

Hide file tree

Showing 2 changed files with 27 additions and 9 deletions.
diff --git a/src/main/java/net/openhft/chronicle/network/cluster/ConnectionManager.java b/src/main/java/net/openhft/chronicle/network/cluster/ConnectionManager.java
@@ -21,20 +21,24 @@
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
-import java.util.IdentityHashMap;
-import java.util.Set;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import static java.util.Collections.newSetFromMap;
-
 public class ConnectionManager<T extends NetworkContext<T>> {
 
     private static final int EMPTY_SEQUENCE = -1;
-    private final Set<ConnectionListenerHolder<?>> connectionListeners = newSetFromMap(new IdentityHashMap<>());
+    private final List<ConnectionListenerHolder<T>> connectionListeners = new ArrayList<>();
     private final AtomicInteger lastListenerAddedSequence = new AtomicInteger(EMPTY_SEQUENCE);
 
     public synchronized void addListener(@NotNull ConnectionListener<T> connectionListener) {
+        //noinspection ForLoopReplaceableByForEach
+        for (int i = 0; i < connectionListeners.size(); i++) {
+            if (connectionListeners.get(i).connectionListener() == connectionListener) {
+                return;
+            }
+        }
         connectionListeners.add(new ConnectionListenerHolder<>(lastListenerAddedSequence.incrementAndGet(), connectionListener));
     }
 
@@ -75,16 +79,18 @@ public EventEmitterToken onConnectionChanged(boolean isConnected,
     private synchronized void executeListenersWithSequenceGreaterThan(int lowerSequenceLimit,
                                                                       @NotNull final T nc,
                                                                       @NotNull EventEmitterToken token) {
-        connectionListeners.forEach(l -> {
+        //noinspection ForLoopReplaceableByForEach
+        for (int i = 0; i < connectionListeners.size(); i++) {
+            ConnectionListenerHolder<T> l = connectionListeners.get(i);
             if (l.sequence > lowerSequenceLimit) {
                 try {
-                    ((ConnectionListenerHolder<T>)l).connectionListener.onConnectionChange(nc, token.connected.get());
+                    l.connectionListener.onConnectionChange(nc, token.connected.get());
                 } catch (IllegalStateException ignore) {
                     // this is already logged
                 }
                 token.latestSequenceExecuted = Math.max(token.latestSequenceExecuted, l.sequence);
             }
-        });
+        }
     }
 
     @FunctionalInterface
@@ -105,10 +111,14 @@ private static final class ConnectionListenerHolder<C extends NetworkContext<C>>
         private final int sequence;
         private final ConnectionListener<C> connectionListener;
 
-        public ConnectionListenerHolder(int sequence, ConnectionListener<C> connectionListener) {
+        public ConnectionListenerHolder(int sequence, @NotNull ConnectionListener<C> connectionListener) {
             this.sequence = sequence;
             this.connectionListener = connectionListener;
         }
+
+        public ConnectionListener<C> connectionListener() {
+            return connectionListener;
+        }
     }
 
     /**

diff --git a/src/test/java/net/openhft/chronicle/network/cluster/ConnectionManagerTest.java b/src/test/java/net/openhft/chronicle/network/cluster/ConnectionManagerTest.java
@@ -130,6 +130,14 @@ void executeNewListenersWorksWhenThereAreNoListeners() {
         verify(listener1).onConnectionChange(networkContext, true);
     }
 
+    @Test
+    void addConnectionListenerIsIdempotent() {
+        connectionManager.addListener(listener1);
+        connectionManager.addListener(listener1);
+        connectionManager.onConnectionChanged(true, networkContext, null);
+        verify(listener1).onConnectionChange(networkContext, true);
+    }
+
     static class TestNetworkContext extends VanillaNetworkContext<TestNetworkContext> {
     }
 }