package metrics_influxdb.measurements;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.util.Collection;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import metrics_influxdb.api.protocols.UDPInfluxdbProtocol;
import metrics_influxdb.misc.Miscellaneous;
import metrics_influxdb.serialization.line.Inliner;

public class UDPInlinerSender extends QueueableSender {
    private final static Logger LOGGER = LoggerFactory.getLogger(UDPInlinerSender.class);
    private static int MAX_MEASURES_IN_SINGLE_POST = 5000;
    private final Inliner inliner;
	private final InetSocketAddress serverAddress;

    public UDPInlinerSender(UDPInfluxdbProtocol protocol) {
        super(MAX_MEASURES_IN_SINGLE_POST);
        inliner = new Inliner();
        serverAddress = new InetSocketAddress(protocol.getHost(), protocol.getPort());
    }

    @Override
    protected boolean doSend(Collection<Measurement> measures) {
        if (measures.isEmpty()) {
            return true;
        }
        
        DatagramChannel channel = null;

		String measuresAsString = inliner.inline(measures);
		try {
			if (LOGGER.isDebugEnabled()) {
            	LOGGER.debug("measurements being sent:\n{}", measuresAsString);
            }
			channel = DatagramChannel.open();
			ByteBuffer buffer = ByteBuffer.wrap(measuresAsString.getBytes(Miscellaneous.UTF8));
			channel.send(buffer, serverAddress);
			LOGGER.debug("{} measurements sent to UDP[{}:{}]", measures.size(), serverAddress.getHostString(), serverAddress.getPort());
			return true;
		} catch (IOException e) {
			LOGGER.info("failed to send {} mesures to UDP[{}:{}], {}", measures.size(), serverAddress.getHostString(), serverAddress.getPort(), e.getMessage(), e);
			return false;
		}
    }
}

@@ -194,16 +194,43 @@
 lection;
+%0Aimport java.util.Iterator;
 %0A%0Aimport
@@ -1158,74 +1158,483 @@
 ll;%0A
-%0A%09%09String measuresAsString = inliner.inline(measures);%0A%09%09try %7B%0A%09%09%09
+        try %7B%0A            channel = DatagramChannel.open();%0A        %7D catch (IOException e) %7B%0A            LOGGER.error(%22failed open udp channel%22, e);%0A            return false;%0A        %7D%0A        Iterator%3CMeasurement%3E measuresIterator = measures.iterator();%0A        int errorCounter = 0;%0A        int successCounter = 0;%0A        while(measuresIterator.hasNext()) %7B%0A            String measuresAsString = inliner.inline(measuresIterator.next());%0A            try %7B%0A                
 if (
@@ -1672,17 +1672,22 @@
         
+     
-%09
+ 
 LOGGER.d
@@ -1703,17 +1703,16 @@
 surement
-s
  being s
@@ -1741,16 +1741,20 @@
 tring);%0A
+    
         
@@ -1763,48 +1763,25 @@
   %7D%0A
-%09%09%09channel = DatagramChannel.open();%0A%09%09%09
+%0A                
 Byte
@@ -1860,19 +1860,32 @@
 UTF8));%0A
-%09%09%09
+                
 channel.
@@ -1913,19 +1913,150 @@
 dress);%0A
-%09%09%09
+                successCounter++;%0A            %7D catch (Throwable e) %7B%0A                errorCounter++;%0A            %7D%0A        %7D%0A        
 LOGGER.d
@@ -2095,18 +2095,63 @@
 P%5B%7B%7D:%7B%7D%5D
-%22,
+; successes: %7B%7D, failures: %7B%7D%22,%0A               
  measure
@@ -2218,29 +2218,92 @@
 rt()
-);%0A%09%09%09return true;%0A%09%09
+, successCounter, errorCounter);%0A        try %7B%0A            channel.close();%0A        
 %7D ca
@@ -2328,22 +2328,32 @@
 ) %7B%0A
-%09%09%09
+            
 LOGGER.
-info
+error
 (%22fa
@@ -2361,160 +2361,76 @@
 led 
-to send %7B%7D mesures to UDP%5B%7B%7D:%7B%7D%5D, %7B%7D%22, measures.size(), serverAddress.getHostString(), serverAddress.getPort(), e.getMessage(), e);%0A%09%09%09return false;%0A%09%09%7D
+close udp channel%22, e);%0A        %7D%0A        return successCounter %3E 0;
 %0A   

send measurements line by line in udp mode