package com.flozano.statsd.netty;

import static java.util.Objects.requireNonNull;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioDatagramChannel;
import io.netty.handler.timeout.WriteTimeoutHandler;

import java.io.Closeable;
import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

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

import com.flozano.statsd.StatsDClient;
import com.flozano.statsd.metrics.Metric;

public class NettyStatsDClientImpl implements StatsDClient, Closeable {

	private static final Logger LOGGER = LoggerFactory
			.getLogger(NettyStatsDClientImpl.class);

	private final Bootstrap bootstrap;
	private final Channel channel;

	private final boolean defaultEventLoopGroup;
	private final EventLoopGroup eventLoopGroup;

	private NettyStatsDClientImpl(String host, int port,
			EventLoopGroup eventLoopGroup, boolean defaultEventLoopGroup,
			int flushProbability) {
		this.eventLoopGroup = requireNonNull(eventLoopGroup);
		this.defaultEventLoopGroup = defaultEventLoopGroup;
		bootstrap = new Bootstrap();
		bootstrap.group(eventLoopGroup);
		bootstrap.channel(NioDatagramChannel.class);
		bootstrap.option(ChannelOption.SO_SNDBUF, 1_000_000);
		bootstrap.option(ChannelOption.SO_RCVBUF, 1_000_000);
		bootstrap.option(ChannelOption.ALLOCATOR, new PooledByteBufAllocator());
		bootstrap.handler(new ChannelInitializer<Channel>() {

			@Override
			protected void initChannel(Channel ch) throws Exception {
				ch.pipeline()
						.addLast("write-timeouts",
								new WriteTimeoutHandler(2, TimeUnit.SECONDS))
						.addLast("error-handler",
								new WriteTimeoutFlushHandler())
						.addLast(
								"udp",
								new BytesToUDPEncoder(host, port,
										flushProbability))
						.addLast("array-encoder",
								new MetricArrayToBytesEncoder(500))
						.addLast("encoder", new MetricToBytesEncoder());
			}
		});
		try {
			this.channel = bootstrap.bind(0).sync().channel();
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}

	}

	public NettyStatsDClientImpl(String host, int port,
			EventLoopGroup eventLoopGroup, int flushProbability) {
		this(host, port, eventLoopGroup, false, flushProbability);
	}

	public NettyStatsDClientImpl(String host, int port, int flushProbability) {
		this(host, port, new NioEventLoopGroup(), true, flushProbability);
	}

	@Override
	public CompletableFuture<Void> send(Metric... metrics) {
		validateMetrics(metrics);

		CompletableFuture<Void> cf = new CompletableFuture<>();
		channel.write(metrics).addListener(
				f -> {
					LOGGER.trace("Message sent (future={}, messages={})", f,
							metrics);
					try {
						f.get();
						if (f.isSuccess()) {
							cf.complete(null);
						} else {
							cf.completeExceptionally(new RuntimeException(
									"Future didn't complete successfully"));
						}
					} catch (ExecutionException e) {
						cf.completeExceptionally(e.getCause());
					}
				});
		return cf;
	}

	@Override
	public void close() throws IOException {
		channel.flush();
		try {
			channel.close().sync();
		} catch (InterruptedException e) {
			LOGGER.warn("Error while closing channel", e);
		} finally {
			if (defaultEventLoopGroup) {
				eventLoopGroup.shutdownGracefully();
			}
		}
	}

	private static void validateMetrics(Metric[] metrics) {
		requireNonNull(metrics);
		if (metrics.length < 1) {
			throw new IllegalArgumentException(
					"At least one metric must be provided");
		}
	}
}

@@ -421,61 +421,8 @@
 nel;
-%0Aimport io.netty.handler.timeout.WriteTimeoutHandler;
 %0A%0Aim
@@ -465,32 +465,84 @@
 io.IOException;%0A
+import java.util.Timer;%0Aimport java.util.TimerTask;%0A
 import java.util
@@ -623,46 +623,8 @@
 ion;
-%0Aimport java.util.concurrent.TimeUnit;
 %0A%0Aim
@@ -1094,16 +1094,50 @@
 Group;%0A%0A
+%09private final Timer flushTimer;%0A%0A
 %09private
@@ -1831,32 +1831,91 @@
 ipeline()%0A%09%09%09%09%09%09
+// Keep original timer-based approach for now.%0A%09%09%09%09%09%09// // 
 .addLast(%22write-
@@ -1923,34 +1923,38 @@
 imeouts%22,%0A%09%09%09%09%09%09
-%09%09
+// // 
 new WriteTimeout
@@ -1981,32 +1981,38 @@
 SECONDS))%0A%09%09%09%09%09%09
+// // 
 .addLast(%22error-
@@ -2023,26 +2023,30 @@
 ler%22,%0A%09%09%09%09%09%09
-%09%09
+// // 
 new WriteTim
@@ -2065,16 +2065,25 @@
 dler())%0A
+%09%09%09%09%09%09//%0A
 %09%09%09%09%09%09.a
@@ -2455,16 +2455,225 @@
 e);%0A%09%09%7D%0A
+%09%09flushTimer = new Timer(%22flush-timer%22);%0A%09%09flushTimer.scheduleAtFixedRate(new TimerTask() %7B%0A%0A%09%09%09@Override%0A%09%09%09public void run() %7B%0A%09%09%09%09channel.flush();%0A%09%09%09%09LOGGER.trace(%22Flushed channel%22);%0A%09%09%09%7D%0A%09%09%7D, 1000, 1000);
 %0A%09%7D%0A%0A%09pu
@@ -3654,16 +3654,39 @@
 ption %7B%0A
+%09%09flushTimer.cancel();%0A
 %09%09channe

Revert to timer-based approach for now.