HCOLL-128 cleanup Replicators closeables handling, fixed small defect -- AbstractConnector.connectLater didn't remove channel from closeables

src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java

@@ -55,7 +55,7 @@ abstract class AbstractChannelReplicator implements Closeable {
     static final int SIZE_OF_SHORT = 2;
     private static final Logger LOG = LoggerFactory.getLogger(AbstractChannelReplicator.class);
     final Selector selector;
-    final Set<Closeable> closeables = Collections.synchronizedSet(new LinkedHashSet<Closeable>());
+    final CloseablesManager closeables = new CloseablesManager();
     private final ExecutorService executorService;
     private final Queue<Runnable> pendingRegistrations = new ConcurrentLinkedQueue<Runnable>();
     @Nullable
@@ -110,28 +110,14 @@ abstract class AbstractChannelReplicator implements Closeable {
 
     @Override
     public void close() {
-        synchronized (this.closeables) {
-            for (Closeable closeable : this.closeables) {
-                try {
-                    closeable.close();
-                } catch (IOException e) {
-                    LOG.error("", e);
-                }
-            }
-            closeables.clear();
-        }
+        closeables.closeQuietly();
         executorService.shutdownNow();
     }
 
     void closeEarlyAndQuietly(SelectableChannel channel) {
-        try {
-            if (throttler != null)
-                throttler.remove(channel);
-            closeables.remove(channel);
-            channel.close();
-        } catch (IOException ex) {
-            // do nothing
-        }
+        if (throttler != null)
+            throttler.remove(channel);
+        closeables.closeQuietly(channel);
     }
 
     /**
@@ -322,20 +308,15 @@ abstract class AbstractChannelReplicator implements Closeable {
          * connectionAttempts}
          */
         public final void connectLater() {
-            try {
-                if (socketChannel != null)
-                    socketChannel.close();
+            if (socketChannel != null) {
+                closeables.closeQuietly(socketChannel);
                 socketChannel = null;
-
-            } catch (IOException e1) {
-                LOG.error("", e1);
             }
 
             final long reconnectionInterval = connectionAttempts * 100;
             if (connectionAttempts < 5)
                 connectionAttempts++;
             doConnect(reconnectionInterval);
-
         }
 
         /**
@@ -354,31 +335,17 @@ abstract class AbstractChannelReplicator implements Closeable {
             final Thread thread = new Thread(new Runnable() {
 
                 public void run() {
-                    SelectableChannel socketChannel = null;
                     try {
                         if (reconnectionInterval > 0)
                             Thread.sleep(reconnectionInterval);
 
-                        synchronized (closeables) {
-                            socketChannel = doConnect();
-                            if (socketChannel == null)
-                                return;
-
-                            closeables.add(socketChannel);
-                            AbstractConnector.this.socketChannel = socketChannel;
-                        }
+                        SelectableChannel socketChannel = doConnect();
+                        closeables.add(socketChannel);
+                        AbstractConnector.this.socketChannel = socketChannel;
 
                     } catch (Exception e) {
-                        if (socketChannel != null)
-                            try {
-                                socketChannel.close();
-                                AbstractConnector.this.socketChannel = null;
-                            } catch (IOException e1) {
-                                //
-                            }
                         LOG.error("", e);
                     }
-
                 }
             });
 

src/main/java/net/openhft/chronicle/map/CloseablesManager.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+final class CloseablesManager implements Closeable {
+    private static final Logger LOG = LoggerFactory.getLogger(TcpReplicator.class.getName());
+
+    private boolean isClosed = false;
+    private List<Closeable> closeables = new ArrayList<Closeable>();
+
+    private void checkState() {
+        if (isClosed)
+            throw new IllegalStateException();
+    }
+
+    public synchronized void add(Closeable closeable) {
+        checkState();
+        if (closeable == null)
+            throw new NullPointerException();
+        closeables.add(closeable);
+    }
+
+    public synchronized void close(Closeable closeable) throws IOException {
+        checkState();
+        for (Iterator<Closeable> iterator = closeables.iterator(); iterator.hasNext(); ) {
+            if (iterator.next() == closeable)
+                iterator.remove();
+        }
+        closeable.close();
+    }
+
+    public synchronized void closeQuietly(Closeable closeable) {
+        try {
+            close(closeable);
+        } catch (IOException e) {
+            LOG.error("", e);
+        }
+    }
+
+    @Override
+    public synchronized void close() throws IOException {
+        IOException ex = null;
+        for (Closeable closeable : closeables) {
+            try {
+                closeable.close();
+            } catch (IOException e) {
+                LOG.error("", e);
+                ex = e;
+            }
+        }
+        closeables.clear();
+        isClosed = true;
+        if (ex != null)
+            throw ex;
+    }
+
+    public synchronized void closeQuietly() {
+        try {
+            close();
+        } catch (IOException e) {
+            // do nothing
+        }
+    }
+}

src/main/java/net/openhft/chronicle/map/TcpReplicator.java

@@ -277,19 +277,13 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
 
         final SocketChannel channel = (SocketChannel) key.channel();
 
-        if (approxTimeOutTime > attached.entryReader.lastHeartBeatReceived + attached.remoteHeartbeatInterval) {
+        if (approxTimeOutTime >
+                attached.entryReader.lastHeartBeatReceived + attached.remoteHeartbeatInterval) {
             if (LOG.isDebugEnabled())
                 LOG.debug("lost connection, attempting to reconnect. " +
                         "missed heartbeat from identifier=" + attached.remoteIdentifier);
-            try {
-                channel.socket().close();
-                channel.close();
-                activeKeys.clear(attached.remoteIdentifier);
-                closeables.remove(channel);
-            } catch (IOException e) {
-                LOG.debug("", e);
-            }
-
+            activeKeys.clear(attached.remoteIdentifier);
+            closeables.closeQuietly(channel.socket());
             attached.connector.connectLater();
         }
     }

src/main/java/net/openhft/chronicle/map/UdpReplicator.java

@@ -169,23 +169,20 @@ class UdpReplicator extends AbstractChannelReplicator implements Replica.Modific
 
         final InetSocketAddress hostAddress = new InetSocketAddress(port);
         client.configureBlocking(false);
-        synchronized (closeables) {
-
-            if (address.isMulticastAddress()) {
-                client.setOption(IP_MULTICAST_IF, networkInterface);
-                client.setOption(SO_REUSEADDR, true);
-                client.bind(hostAddress);
-                client.join(address, networkInterface);
-                if (LOG.isDebugEnabled())
-                    LOG.debug("Connecting via multicast, group=" + address);
-            } else {
-                client.bind(hostAddress);
-            }
-
+        if (address.isMulticastAddress()) {
+            client.setOption(IP_MULTICAST_IF, networkInterface);
+            client.setOption(SO_REUSEADDR, true);
+            client.bind(hostAddress);
+            client.join(address, networkInterface);
             if (LOG.isDebugEnabled())
-                LOG.debug("Listening on port " + port);
-            closeables.add(client);
+                LOG.debug("Connecting via multicast, group=" + address);
+        } else {
+            client.bind(hostAddress);
         }
+
+        if (LOG.isDebugEnabled())
+            LOG.debug("Listening on port " + port);
+        closeables.add(client);
         return client;
     }