package com.alibaba.otter.canal.parse.driver.mysql.socket;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.AdaptiveRecvByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.io.IOException;
import java.net.SocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

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

/**
 * @author luoyaogui 实现channel的管理（监听连接、读数据、回收） 2016-12-28
 */
@SuppressWarnings({ "rawtypes", "deprecation" })
public abstract class NettySocketChannelPool {

    private static EventLoopGroup              group     = new NioEventLoopGroup();                              // 非阻塞IO线程组
    private static Bootstrap                   boot      = new Bootstrap();                                      // 主
    private static Map<Channel, SocketChannel> chManager = new ConcurrentHashMap<>();
    private static final Logger                logger    = LoggerFactory.getLogger(NettySocketChannelPool.class);

    static {
        boot.group(group)
            .channel(NioSocketChannel.class)
            .option(ChannelOption.TCP_NODELAY, true)
            // 如果是延时敏感型应用，建议关闭Nagle算法
            .option(ChannelOption.SO_KEEPALIVE, true)
            .option(ChannelOption.SO_REUSEADDR, true)
            .option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)
            .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            //
            .handler(new ChannelInitializer() {

                @Override
                protected void initChannel(Channel ch) throws Exception {
                    ch.pipeline().addLast(new BusinessHandler());// 命令过滤和handler添加管理
                }
            });
    }

    public static SocketChannel open(SocketAddress address) throws Exception {
        SocketChannel socket = null;
        ChannelFuture future = boot.connect(address).sync();

        if (future.isSuccess()) {
            future.channel().pipeline().get(BusinessHandler.class).latch.await();
            socket = chManager.get(future.channel());
        }

        if (null == socket) {
            throw new IOException("can't create socket!");
        }

        return socket;
    }

    public static class BusinessHandler extends SimpleChannelInboundHandler<ByteBuf> {

        private NettySocketChannel   socket = null;
        private final CountDownLatch latch  = new CountDownLatch(1);

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            socket.setChannel(null);
            chManager.remove(ctx.channel());// 移除
            super.channelInactive(ctx);
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            socket = new NettySocketChannel();
            socket.setChannel(ctx.channel());
            chManager.put(ctx.channel(), socket);
            latch.countDown();
            super.channelActive(ctx);
        }

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
            if (socket != null) {
                socket.writeCache(msg);
            } else {
                // TODO: need graceful error handler.
                logger.error("no socket available.");
            }
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            // need output error for troubeshooting.
            logger.error("business error.", cause);
            ctx.close();
        }
    }
}
