public final class CandidateSockets
{
    private static final float AGRONA_DEFAULT_LOAD_FACTOR = 0.67f;

    private final AtomicReference<Long2ObjectHashMap<InetSocketAddress>> candidateSockets =
            new AtomicReference<>(new Long2ObjectHashMap<>());

    public void beginMonitoringSocketIdentifier(final InetSocketAddress socketAddress, final long socketIdentifier)
    {
        while (true)
        {
            final Long2ObjectHashMap<InetSocketAddress> candidateSnapshot = candidateSockets.get();
            final Long2ObjectHashMap<InetSocketAddress> updated = new Long2ObjectHashMap<>(candidateSnapshot.size(), AGRONA_DEFAULT_LOAD_FACTOR);

            final Long2ObjectHashMap<InetSocketAddress>.KeyIterator keyIterator = candidateSnapshot.keySet().iterator();
            while(keyIterator.hasNext())
            {
                final long key = keyIterator.nextLong();
                updated.put(key, candidateSnapshot.get(key));
            }

            updated.put(socketIdentifier, socketAddress);

            if (candidateSockets.compareAndSet(candidateSnapshot, updated))
            {
                break;
            }
        }
    }

    public void endMonitoringOfSocketIdentifier(final long socketIdentifier)
    {
        while (true)
        {
            final Long2ObjectHashMap<InetSocketAddress> candidateSnapshot = candidateSockets.get();
            final Long2ObjectHashMap<InetSocketAddress> updated = new Long2ObjectHashMap<>(candidateSnapshot.size(), AGRONA_DEFAULT_LOAD_FACTOR);

            final Long2ObjectHashMap<InetSocketAddress>.KeyIterator keyIterator = candidateSnapshot.keySet().iterator();
            while(keyIterator.hasNext())
            {
                final long key = keyIterator.nextLong();
                if(key != socketIdentifier)
                {
                    updated.put(key, candidateSnapshot.get(key));
                }
            }

            if (candidateSockets.compareAndSet(candidateSnapshot, updated))
            {
                break;
            }
        }
    }

    public Long2ObjectHashMap<InetSocketAddress> getSnapshot()
    {
        return candidateSockets.get();
    }
} public final class MonitoredSockets<T extends DescribableSocket>
{
    private final Long2ObjectHashMap<T> monitoredSocketInstances = new Long2ObjectHashMap<>();
    private final SocketMonitoringLifecycleListener lifecycleListener;
    private final SocketDescriptor socketDescriptor = new SocketDescriptor();
    private final LongHashSet keysForRemoval = new LongHashSet();

    public MonitoredSockets(final SocketMonitoringLifecycleListener lifecycleListener)
    {
        this.lifecycleListener = lifecycleListener;
    }

    public boolean contains(final long socketInstanceIndentifier)
    {
        return monitoredSocketInstances.containsKey(socketInstanceIndentifier);
    }

    public void put(final long socketInstanceIndentifier, final T entry)
    {
        monitoredSocketInstances.put(socketInstanceIndentifier, entry);
        entry.describeTo(socketDescriptor);

        lifecycleListener.socketMonitoringStarted(
                socketDescriptor.getAddress(), socketDescriptor.getPort(), socketDescriptor.getInode());
    }

    public T get(final long socketInstanceIndentifier)
    {
        return monitoredSocketInstances.get(socketInstanceIndentifier);
    }

    public void purgeEntriesOlderThan(final long latestUpdateCount)
    {
        keysForRemoval.clear();
        final Long2ObjectHashMap<T>.KeyIterator iterator = monitoredSocketInstances.keySet().iterator();
        while(iterator.hasNext())
        {
            final long key = iterator.nextLong();
            if(monitoredSocketInstances.get(key).getUpdateCount() != latestUpdateCount)
            {
                keysForRemoval.add(key);
            }
        }

        final LongIterator keyIterator = keysForRemoval.iterator();
        while(keyIterator.hasNext())
        {
            final long key = keyIterator.nextValue();
            final T staleEntry = monitoredSocketInstances.remove(key);
            staleEntry.describeTo(socketDescriptor);
            lifecycleListener.socketMonitoringStopped(
                    socketDescriptor.getAddress(), socketDescriptor.getPort(), socketDescriptor.getInode());
        }
    }
} public final class SocketIdentifier
{
    private SocketIdentifier() {}

    /**
     * Pack IPv4 address and socket into a long.
     * @param socketAddress the socket address
     * @return the encoded value
     */
    public static long fromInet4SocketAddress(final InetSocketAddress socketAddress)
    {
        validateAddressType(socketAddress);
        final long ipAddressOctets = Integer.toUnsignedLong(socketAddress.getAddress().hashCode());
        final long port = socketAddress.getPort();
        return port << 32 | ipAddressOctets;
    }

    /**
     * Pack IPv4 address and match-all socket flag into a long.
     * @param inetAddress the host address
     * @return the encoded value
     */
    public static long fromInet4Address(final InetAddress inetAddress)
    {
        ensureIsInet4Address(inetAddress);

        return Integer.toUnsignedLong(inetAddress.hashCode());
    }

    /**
     * Pack IPv4 address and socket into a long.
     * @param socketAddress the socket address
     * @param inode the socket inode
     * @return the encoded value
     */
    public static long fromInet4SocketAddressAndInode(final InetSocketAddress socketAddress, final int inode)
    {
        return overlayInode(fromInet4SocketAddress(socketAddress), inode);
    }

    /**
     * Pack IPv4 address and match-all socket flag into a long.
     * @param inetAddress the host address
     * @param inode the socket inode
     * @return the encoded value
     */
    public static long fromInet4AddressAndInode(final InetAddress inetAddress, final int inode)
    {
        return overlayInode(fromInet4Address(inetAddress), inode);
    }

    /**
     * Is this socketIdentifier a match for all ports on the encoded IP address.
     * @param socketIdentifier the encoded value
     * @return whether this socketIdentifier matches all ports
     */
    public static boolean isMatchAllSocketFlagSet(final long socketIdentifier)
    {
        return extractPortNumber(socketIdentifier) == 0L;
    }

    /**
     * Mast the port number out of this socket identifier.
     * @param socketIdentifier the encoded value
     * @return the socketIdentifier with the port number masked out
     */
    public static long asMatchAllSocketsSocketIdentifier(final long socketIdentifier)
    {
        return 0xFFFF0000FFFFFFFFL & socketIdentifier;
    }

    /**
     * Pack a hex-encoded IPv4:port socket address into a long.
     * @param decodedAddress 32-bit IPv4 address decoded from hex
     * @param port the port number
     * @return the encoded value
     */
    public static long fromLinuxKernelHexEncodedAddressAndPort(final long decodedAddress, final long port)
    {
        return port << 32 | Long.reverseBytes(decodedAddress) >>> 32;
    }

    /**
     * Pack inode value into socketIdentifier to differentiate between difference sockets listening to the same port.
     * @param socketIdentifier the socketIdentifier
     * @param inode the inode
     * @return the encoded value
     */
    public static long overlayInode(final long socketIdentifier, final long inode)
    {
        return  inode << 48 | socketIdentifier;
    }

    /**
     * Extract port number from a socketIdentifier.
     * @param socketIdentifier the socketIdentifier
     * @return the port number
     */
    public static int extractPortNumber(final long socketIdentifier)
    {
        return (int) ((socketIdentifier >> 32) & 0xFFFFL);
    }

    /**
     * Generates a vaguely human-readable format for a given socket identifier.
     * @param socketIdentifier the encoded socket identifier
     * @return human-readable form
     */
    public static String toDebugString(final long socketIdentifier)
    {
        final int ipBits = (int) socketIdentifier;
        final int port = extractPortNumber(socketIdentifier);
        final int inode = (int) (socketIdentifier >> 48);

        return Integer.toHexString(ipBits) + ":" + port + "/" + inode;
    }

    private static void validateAddressType(final InetSocketAddress socketAddress)
    {
        ensureIsInet4Address(socketAddress.getAddress());
    }

    private static void ensureIsInet4Address(final InetAddress address)
    {
        if(!(address instanceof Inet4Address))
        {
            throw new IllegalArgumentException("Due to the nature of some awful hacks, " +
                    "I only work with Inet4Address-based sockets, not: " + address.getClass().getSimpleName());
        }
    }
} public interface SocketMonitoringLifecycleListener
{
    /**
     * Socket is available for monitoring.
     * @param inetAddress the ip address
     * @param port the port
     * @param inode the inode of the socket
     */
    void socketMonitoringStarted(final InetAddress inetAddress, final int port, final long inode);

    /**
     * Socket is unavailable for monitoring.
     * @param inetAddress the ip address
     * @param port the port
     * @param inode the inode of the socket
     */
    void socketMonitoringStopped(final InetAddress inetAddress, final int port, final long inode);
} public final class FileLoader
{
    private final Path path;
    private final ByteBuffer tmp = ByteBuffer.allocateDirect(4096);
    private ByteBuffer buffer;
    private FileChannel fileChannel;

    public FileLoader(final Path path, final int initialBufferCapacity)
    {
        this.path = path;
        buffer = ByteBuffer.allocateDirect(initialBufferCapacity);
    }

    /**
     * Opens a channel to the specified path if it does not already exist.
     * Allocates a larger ByteBuffer if file size &gt; current buffer size.
     * Reads file data into ByteBuffer.
     */
    public void load()
    {
        try
        {
            if (fileChannel == null)
            {
                fileChannel = FileChannel.open(path, StandardOpenOption.READ);
            }

            fileChannel.position(0L);
            buffer.clear();
            tmp.clear();
            while(fileChannel.read(tmp) > 0)
            {
                tmp.flip();
                if(tmp.remaining() > buffer.capacity() - buffer.position())
                {
                    final ByteBuffer next = ByteBuffer.allocateDirect(Math.max(buffer.capacity() * 2, tmp.remaining()));
                    buffer.flip();
                    next.put(buffer);
                    buffer = next;
                }
                buffer.put(tmp);
                tmp.clear();
            }

            buffer.flip();
        }
        catch(final IOException e)
        {
            throw new UncheckedIOException(e);
        }
    }

    public ByteBuffer getBuffer()
    {
        return buffer;
    }
} public final class Parsers
{
    private static final byte COLUMN_DELIMITER = (byte) ' ';
    private static final byte ROW_DELIMITER = (byte) '\n';

    public static TokenHandler rowColumnParser(
            TokenHandler tokenHandler)
    {
        return new DelimitedDataParser(
                new DelimitedDataParser(
                        tokenHandler,
                        COLUMN_DELIMITER,
                        true),
                ROW_DELIMITER,
                true);
    }

    private Parsers() {}
} public interface TokenHandler
{
    /**
     * Handle a single token.
     * @param src data source
     * @param startPosition the start position of the token
     * @param endPosition the end position of the token
     */
    void handleToken(final ByteBuffer src, final int startPosition, final int endPosition);

    /**
     * Current token set is complete.
     */
    void complete();

    /**
     * Reset state in preparation for handling a new data set.
     */
    void reset();
} public static long asMatchAllSocketsSocketIdentifier(final long socketIdentifier)
{
    return 0xFFFF0000FFFFFFFFL & socketIdentifier;
} public static long fromInet4AddressAndInode(final InetAddress inetAddress, final int inode)
{
    return overlayInode(fromInet4Address(inetAddress), inode);
} public static long fromInet4SocketAddressAndInode(final InetSocketAddress socketAddress, final int inode)
{
    return overlayInode(fromInet4SocketAddress(socketAddress), inode);
}
import com.lmax.angler.monitoring.network.monitor.socket.CandidateSockets;
import com.lmax.angler.monitoring.network.monitor.socket.MonitoredSockets;
import com.lmax.angler.monitoring.network.monitor.socket.SocketIdentifier;
import com.lmax.angler.monitoring.network.monitor.socket.SocketMonitoringLifecycleListener;
import com.lmax.angler.monitoring.network.monitor.util.FileLoader;
import com.lmax.angler.monitoring.network.monitor.util.Parsers;
import com.lmax.angler.monitoring.network.monitor.util.TokenHandler;
import org.agrona.collections.Long2ObjectHashMap;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.file.Path;
import java.nio.file.Paths;
import static com.lmax.angler.monitoring.network.monitor.socket.SocketIdentifier.asMatchAllSocketsSocketIdentifier;
import static com.lmax.angler.monitoring.network.monitor.socket.SocketIdentifier.fromInet4AddressAndInode;
import static com.lmax.angler.monitoring.network.monitor.socket.SocketIdentifier.fromInet4SocketAddressAndInode;
package com.lmax.angler.monitoring.network.monitor.socket.tcp;




/**
 * Monitor for reporting changes in /proc/net/tcp.
 */
public final class TcpSocketMonitor
{
    private final MonitoredSockets<TcpBufferStats> monitoredSockets;
    private final CandidateSockets candidateSockets = new CandidateSockets();
    private final TcpColumnHandler tokenHandler = new TcpColumnHandler(this::handleEntry);
    private final TokenHandler lineParser = Parsers.rowColumnParser(tokenHandler);