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.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timer;

import java.io.Closeable;
import java.io.IOException;
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 Timer flushTimer;

	private final boolean defaultEventLoopGroup;
	private final EventLoopGroup eventLoopGroup;

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

			@Override
			protected void initChannel(Channel ch) throws Exception {
				ch.pipeline()
						.addFirst("encoder", new MetricToBytesEncoder())
						.addLast(
								"logging-udp",
								new LoggingHandler(NettyStatsDClientImpl.class,
										LogLevel.DEBUG))
						.addLast("udp", new BytesToUDPEncoder(host, port));
			}
		});
		try {
			this.channel = bootstrap.bind(0).sync().channel();
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}

		flushTimer = new HashedWheelTimer();
		flushTimer.newTimeout(timeout -> channel.flush(), 1, TimeUnit.SECONDS);
	}

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

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

	@Override
	public void send(Metric... metrics) {
		for (Metric m : metrics) {
			channel.write(m).addListener(
					f -> LOGGER.trace("Message sent (future={}, message={})",
							f, m));
		}
	}

	@Override
	public void close() throws IOException {
		flushTimer.stop();
		channel.flush();
		if (defaultEventLoopGroup) {
			eventLoopGroup.shutdownGracefully();
		}
	}
}

@@ -117,55 +117,8 @@
 ap;%0A
-import io.netty.buffer.PooledByteBufAllocator;%0A
 impo
@@ -138,32 +138,32 @@
 hannel.Channel;%0A
+
 import io.netty.
@@ -194,47 +194,8 @@
 er;%0A
-import io.netty.channel.ChannelOption;%0A
 impo
@@ -336,98 +336,8 @@
 el;%0A
-import io.netty.handler.logging.LogLevel;%0Aimport io.netty.handler.logging.LoggingHandler;%0A
 impo
@@ -1341,24 +1341,27 @@
 l.class);%0A%09%09
+// 
 bootstrap.op
@@ -1393,16 +1393,21 @@
 TOR, new
+%0A%09%09//
  PooledB
@@ -1578,240 +1578,110 @@
 ne()
-%0A%09%09%09%09%09%09
 .add
-Fir
+La
 st(%22
-encoder%22, new MetricToBytesEncoder())%0A%09%09%09%09%09%09.addLast(%0A%09%09%09%09%09%09%09%09%22logging-udp%22,%0A%09%09%09%09%09%09%09%09new LoggingHandler(NettyStatsDClientImpl.class,%0A%09%09%09%09%09%09%09%09%09%09LogLevel.DEBUG))%0A%09%09%09%09%09%09.addLast(%22udp%22, new BytesToUDPEncoder(host, port
+udp%22, new BytesToUDPEncoder(host, port))%0A%09%09%09%09%09%09.addLast(%22encoder%22, new MetricToBytesEncoder(
 ));%0A
@@ -2307,19 +2307,32 @@
 ner(
-%0A%09%09%09%09%09f -%3E 
+f -%3E %7B%0A%09%09%09%09f.get();%0A%09%09%09%09
 LOGG
@@ -2383,21 +2383,20 @@
 %7D)%22,
-%0A%09%09%09%09%09%09%09f, m)
+ f, m);%0A%09%09%09%7D
 );%0A%09

Fix order of handlers