package net.openhft.chronicle.network.cluster;

import net.openhft.chronicle.core.io.AbstractCloseable;
import net.openhft.chronicle.core.io.Closeable;
import net.openhft.chronicle.core.threads.EventHandler;
import net.openhft.chronicle.core.threads.EventLoop;
import net.openhft.chronicle.core.threads.HandlerPriority;
import net.openhft.chronicle.core.threads.InvalidEventHandlerException;
import net.openhft.chronicle.network.NetworkStatsListener;
import net.openhft.chronicle.network.TCPRegistry;
import net.openhft.chronicle.network.TcpEventHandler;
import net.openhft.chronicle.network.tcp.ChronicleServerSocket;
import net.openhft.chronicle.network.tcp.ChronicleServerSocketChannel;
import net.openhft.chronicle.network.tcp.ChronicleSocketChannel;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.ClosedChannelException;
import java.util.Collection;
import java.util.concurrent.CopyOnWriteArrayList;

import static net.openhft.chronicle.core.io.Closeable.closeQuietly;
import static net.openhft.chronicle.network.NetworkStatsListener.notifyHostPort;

public class ClusterAcceptorEventHandler<C extends ClusterContext<C, T>, T extends ClusteredNetworkContext<T>> extends AbstractCloseable implements EventHandler {
    private static final Logger LOGGER = LoggerFactory.getLogger(ClusterAcceptorEventHandler.class);
    @NotNull
    private final ChronicleServerSocketChannel ssc;
    @NotNull
    private final C context;
    private final String hostPort;
    private final Collection<Closeable> closeables = new CopyOnWriteArrayList<>();
    private EventLoop eventLoop;

    public ClusterAcceptorEventHandler(@NotNull final String hostPort,
                                       @NotNull final C context) throws IOException {
        this.hostPort = hostPort;
        this.ssc = TCPRegistry.acquireServerSocketChannel(this.hostPort);
        this.context = context;
    }

    @Override
    public void eventLoop(final EventLoop eventLoop) {
        this.eventLoop = eventLoop;
    }

    @Override
    public boolean action() throws InvalidEventHandlerException {
        if (!ssc.isOpen() || isClosed() || eventLoop.isClosed())
            throw new InvalidEventHandlerException();

        try {
            LOGGER.debug("accepting {}", ssc);

            final ChronicleSocketChannel sc = ssc.accept();

            if (sc != null) {
                if (isClosed() || eventLoop.isClosed()) {
                    closeQuietly(sc);
                    throw new InvalidEventHandlerException("closed");
                }
                final T nc = context.networkContextFactory().apply(context);
                closeables.add(nc);
                nc.socketChannel(sc);
                nc.isAcceptor(true);
                final NetworkStatsListener<T> nl = nc.networkStatsListener();
                notifyHostPort(sc, nl);
                final TcpEventHandler<T> tcpEventHandler = context.tcpEventHandlerFactory().apply(nc);

                if (isClosed())
                    closeQuietly(nc);
                else
                    eventLoop.addHandler(tcpEventHandler);
            }
        } catch (AsynchronousCloseException e) {
            closeSocket();
            throw new InvalidEventHandlerException(e);
        } catch (ClosedChannelException e) {
            closeSocket();
            if (isClosed())
                throw InvalidEventHandlerException.reusable();
            else
                throw new InvalidEventHandlerException(e);
        } catch (Exception e) {
            if (!isClosed() && !eventLoop.isClosed()) {
                final ChronicleServerSocket socket = ssc.socket();
                LOGGER.warn("{}, port={}", hostPort, socket == null ? "unknown" : socket.getLocalPort(), e);
            }
            closeSocket();
            throw new InvalidEventHandlerException(e);
        }
        return false;
    }

    private void closeSocket() {
        closeQuietly(ssc);
    }

    @NotNull
    @Override
    public HandlerPriority priority() {
        return HandlerPriority.BLOCKING;
    }

    @Override
    protected void performClose() {
        closeQuietly(closeables);
        closeables.clear(); // https://github.com/OpenHFT/Chronicle-Network/issues/104
        closeSocket();
    }
}

@@ -101,56 +101,8 @@
 le;%0A
-import net.openhft.chronicle.core.io.Closeable;%0A
 impo
@@ -145,32 +145,32 @@
 s.EventHandler;%0A
+
 import net.openh
@@ -933,87 +933,8 @@
 ion;
-%0Aimport java.util.Collection;%0Aimport java.util.concurrent.CopyOnWriteArrayList;
 %0A%0Aim
@@ -1491,91 +1491,8 @@
 rt;%0A
-    private final Collection%3CCloseable%3E closeables = new CopyOnWriteArrayList%3C%3E();%0A
     
@@ -2555,44 +2555,8 @@
 t);%0A
-                closeables.add(nc);%0A
     
@@ -2913,18 +2913,31 @@
 Quietly(
-nc
+tcpEventHandler
 );%0A     
@@ -3986,24 +3986,24 @@
   @Override%0A
+
     protecte
@@ -4030,129 +4030,8 @@
 ) %7B%0A
-        closeQuietly(closeables);%0A        closeables.clear(); // https://github.com/OpenHFT/Chronicle-Network/issues/104%0A
     

Stop accumulating NetworkContexts in ClusterAcceptorEventHandler, Fixes #115