package org.collectd.services;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.MulticastSocket;
import java.net.SocketException;
import java.util.Arrays;
import lombok.extern.slf4j.Slf4j;
import org.collectd.config.CollectdConstants;
import org.collectd.model.Notification;
import org.collectd.model.PluginData;
import org.collectd.model.Values;
import org.collectd.protocol.UdpBufferWriter;

/**
 * UDP packet writer for Collectd.
 */
@Slf4j
public class UdpPacketSender {

    private UdpBufferWriter writer;

    private final InetSocketAddress server;

    private DatagramSocket socket;
    private MulticastSocket mcast;
    
    private String client;

    /**
     * Create new UDP packet writer instance. Default packet size is used.
     *
     * @param server Collectd server address
     */
    public UdpPacketSender(final InetSocketAddress server) {
        this(server, null, CollectdConstants.DEFAULT_PACKET_SIZE);
    }

    /**
     * Create new UDP packet writer instance.
     *
     * @param server Collectd server address
     * @param clientHost client hostname
     * @param packetSize packet size
     */
    public UdpPacketSender(final InetSocketAddress server, final String clientHost, final int packetSize) {
        this.server = server;
        this.client = clientHost;
        writer = new UdpBufferWriter(packetSize);
    }

    /**
     * Write value list. Signature and encrypted parts are not supported yet.
     *
     * @param values numeric value list
     * @throws IOException unable to write value to output stream
     */
    public void send(final Values values) throws IOException {
        setDefaults(values);
        
        final byte[] bufferToSend = writer.checkSpace(values);
        if (bufferToSend != null) {
            flush(bufferToSend);
        }
        
        writer.writeValuesPart(values);
    }

    /**
     * Write notification. Signature and encrypted parts are not supported yet.
     *
     * @param notification notification
     * @throws IOException unable to write value to output stream
     */
    public void send(final Notification notification) throws IOException {
        setDefaults(notification);
        
        final byte[] bufferToSend = writer.checkSpace(notification);
        if (bufferToSend != null) {
            flush(bufferToSend);
        }
        
        writer.writeNotificationPart(notification);
    }

    public void flush() throws IOException {
        flush(writer.getBuffer());
    }

    private String getClient() {
        if (client == null) {
            try {
                client = InetAddress.getLocalHost().getHostName();
            } catch (IOException ex) {
                log.error("Unable to get host name", ex);
                client = "unknown";
            }
        }
        return client;
    }

    private void setDefaults(final PluginData data) {
        if (data.getHost() == null) {
            data.setHost(getClient());
        }
        if (data.getTime() <= 0) {
            data.setTime(System.currentTimeMillis());
        }
    }

    private DatagramSocket getSocket() throws SocketException {
        if (socket == null) {
            socket = new DatagramSocket();
        }
        return socket;
    }

    private MulticastSocket getMulticastSocket() throws IOException {
        if (mcast == null) {
            mcast = new MulticastSocket();
            mcast.setTimeToLive(1);
        }
        return mcast;
    }
    
    private void flush(final byte[] buffer) throws IOException {
        final int length = buffer.length;
        if (length == 0) {
            return;
        }

        if (log.isDebugEnabled()) {
            log.debug("Sending UDP packet, buffer length: " + length);
        }
        if (log.isTraceEnabled()) {
            log.trace("Destination host: " + server.getHostString());
            log.trace("Destination port: " + server.getPort());
            log.trace("Buffer data: " + Arrays.toString(buffer));
        }

        final DatagramPacket packet = new DatagramPacket(buffer, length, server);
        if (server.getAddress().isMulticastAddress()) {
            getMulticastSocket().send(packet);
        } else {
            getSocket().send(packet);
        }
    }
}

@@ -272,16 +272,113 @@
 Arrays;%0A
+import java.util.concurrent.atomic.AtomicInteger;%0Aimport java.util.concurrent.atomic.AtomicLong;%0A
 import l
@@ -403,16 +403,16 @@
 .Slf4j;%0A
-
 import o
@@ -842,21 +842,17 @@
  mcast;%0A
-    
 %0A
+
     priv
@@ -867,24 +867,154 @@
 ng client;%0A%0A
+    private final AtomicLong bytesSent = new AtomicLong(0L);%0A    private final AtomicInteger packetsSent = new AtomicInteger(0);%0A%0A
     /**%0A    
@@ -1999,25 +1999,17 @@
 alues);%0A
-        
 %0A
+
         
@@ -2138,33 +2138,25 @@
 ;%0A        %7D%0A
-        
 %0A
+
         writ
@@ -2509,25 +2509,17 @@
 ation);%0A
-        
 %0A
+
         
@@ -2662,17 +2662,9 @@
   %7D%0A
-        
 %0A
+
     
@@ -2797,32 +2797,190 @@
 ffer());%0A    %7D%0A%0A
+    public long getBytesSent() %7B%0A        return bytesSent.longValue();%0A    %7D%0A%0A    public int getPacketsSent() %7B%0A        return packetsSent.intValue();%0A    %7D%0A%0A
     private Stri
@@ -3942,13 +3942,9 @@
   %7D%0A
-    
 %0A
+
     
@@ -4093,16 +4093,16 @@
 return;%0A
-
         
@@ -4096,32 +4096,109 @@
 urn;%0A        %7D%0A%0A
+        bytesSent.addAndGet(length);%0A        packetsSent.incrementAndGet();%0A%0A
         if (log.

add stats counters