package com.lmax.angler.monitoring.network.monitor.app;

import com.lmax.angler.monitoring.network.monitor.socket.SocketMonitoringLifecycleListener;
import com.lmax.angler.monitoring.network.monitor.socket.tcp.TcpSocketMonitor;
import com.lmax.angler.monitoring.network.monitor.socket.tcp.TcpSocketStatisticsHandler;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public final class ReportTcpSocketQueueDepthMain implements
        SocketMonitoringLifecycleListener, TcpSocketStatisticsHandler, Runnable
{
    private static final long SAMPLE_INTERVAL_MS = Long.getLong("angler.tcp.sample.ms", 1000);
    private final TcpSocketMonitor tcpSocketMonitor;

    public static void main(String[] args) throws UnknownHostException
    {
        if (args.length == 0)
        {
            throw new IllegalStateException("Supply an address to monitor (e.g 10.0.7.24)");
        }
        new ReportTcpSocketQueueDepthMain(InetAddress.getByName(args[0])).start();
    }

    private ReportTcpSocketQueueDepthMain(final InetAddress addressToMonitor)
    {
        tcpSocketMonitor = new TcpSocketMonitor(this);
        tcpSocketMonitor.beginMonitoringOf(addressToMonitor);
    }

    private void start()
    {
        Executors.newSingleThreadScheduledExecutor().
                scheduleAtFixedRate(this, SAMPLE_INTERVAL_MS, SAMPLE_INTERVAL_MS, TimeUnit.MILLISECONDS);
    }

    @Override
    public void run()
    {
        tcpSocketMonitor.poll(this);
    }

    @Override
    public void onStatisticsUpdated(final InetAddress inetAddress, final int port,
                                    final long socketIdentifier, final long inode,
                                    final long receiveQueueDepth, final long transmitQueueDepth)
    {
        System.out.printf("%d %s:%d; RX: %d, TX: %d%n", TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis()),
                inetAddress, port, receiveQueueDepth, transmitQueueDepth);
    }

    @Override
    public void socketMonitoringStarted(final InetAddress inetAddress, final int port, final long inode)
    {
        System.out.printf("Started monitoring %s:%d%n", inetAddress, port);
    }

    @Override
    public void socketMonitoringStopped(InetAddress inetAddress, int port, long inode)
    {
        // no-op
    }
}
@@ -340,16 +340,51 @@
 ddress;%0A
+import java.net.InetSocketAddress;%0A
 import j
@@ -413,16 +413,16 @@
 eption;%0A
-
 import j
@@ -489,16 +489,64 @@
 imeUnit;
+%0Aimport java.util.HashMap;%0Aimport java.util.Map;
 %0A%0Apublic
@@ -828,16 +828,204 @@
 Monitor;
+%0A    private final Map%3CInetSocketAddress, Long%3E rxMaxima = new HashMap%3C%3E();%0A    private final Map%3CInetSocketAddress, Long%3E txMaxima = new HashMap%3C%3E();%0A    private long nextReportTimestamp;
 %0A%0A    pu
@@ -2115,55 +2115,305 @@
     
-System.out.printf(%22%25d %25s:%25d; RX: %25d, TX: %25d%25n%22,
+final InetSocketAddress key = new InetSocketAddress(inetAddress, port);%0A        updateMaximum(receiveQueueDepth, rxMaxima, key);%0A        updateMaximum(transmitQueueDepth, txMaxima, key);%0A%0A        if (System.currentTimeMillis() %3E nextReportTimestamp)%0A        %7B%0A            final long epochSeconds =
  Tim
@@ -2468,17 +2468,17 @@
 illis())
-,
+;
 %0A       
@@ -2486,68 +2486,704 @@
     
-    inetAddress, port, receiveQueueDepth, transmitQueueDepth
+for (final InetSocketAddress socket : rxMaxima.keySet())%0A            %7B%0A                System.out.printf(%22%25d %25s; RX: %25s, TX: %25s%25n%22, epochSeconds,%0A                  socket, rxMaxima.get(socket), txMaxima.get(socket));%0A            %7D%0A            nextReportTimestamp = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(1L);%0A            rxMaxima.clear();%0A            txMaxima.clear();%0A        %7D%0A    %7D%0A%0A    private void updateMaximum(final long value, final Map%3CInetSocketAddress, Long%3E maxima, final InetSocketAddress key)%0A    %7B%0A        Long existing = maxima.get(key);%0A        if (existing == null)%0A        %7B%0A            existing = 0L;%0A        %7D%0A        maxima.put(key, Math.max(existing, value)
 );%0A 

Make tcp sample rate configurable