package com.twjitm.core.common.service.rpc.network;

import com.twjitm.core.common.netstack.entity.rpc.NettyRpcRequestMessage;
import com.twjitm.core.common.service.rpc.server.NettyRpcNodeInfo;
import com.twjitm.core.utils.logs.LoggerUtils;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.log4j.Logger;

import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.locks.ReentrantLock;

/**
 * rpc消息连接实体，主要是讲封装好的请求包，通过连接对象，启动一个netty本地服务器，
 * 然后通过服务器连接到远程服务器。
 *
 * @author twjitm - [Created on 2018-08-20 11:01]
 * @company https://github.com/twjitm/
 * @jdk java version "1.8.0_77"
 * rpc 连接实体对象
 */
public class NettyRpcClientConnection {
    private Logger logger = LoggerUtils.getLogger(NettyRpcClientConnection.class);

    private NioSocketChannel channel;

    private ReentrantLock statusLock;
    /**
     * 重连线程池工具
     */
    private ExecutorService executorService;
    private EventLoopGroup eventLoopGroup = new NioEventLoopGroup(1);
    /**
     * 是否启用重连
     */
    private volatile boolean reConnectOn = true;

    private NettyRpcClient nettyRpcClient;
    private NettyRpcNodeInfo nettyRpcNodeInfo;

    public NettyRpcClientConnection(NettyRpcClient nettyRpcClient, NettyRpcNodeInfo nettyRpcNodeInfo, ExecutorService executorService) {
        if (executorService == null) {
            throw new IllegalArgumentException("ALL PARAMETERS MUST ACCURATE.");
        }
        this.nettyRpcClient = nettyRpcClient;
        this.nettyRpcNodeInfo = nettyRpcNodeInfo;
        this.executorService = executorService;
        this.statusLock = new ReentrantLock();
    }

    /**
     * 创建打开连接，此方法很重要
     * 所谓打开连接，其实相当于启动netty客户端程序一样
     * 将启动程序封装到NettyRpcServerConnectTask类中，可以看到
     * 当提交一个NettyRpcServerConnectTask任务时候，利用java
     * 提供的Future类来提交一个任务，我们可以看到这个submit是一个
     * 同步阻塞试方法。
     *
     * @return
     */
    public boolean open() {
        // 判断是否已经连接
        if (isConnected()) {
            throw new IllegalStateException("ALREADY CONNECTED. DISCONNECT FIRST.");
        }
        // 创建Socket连接
        try {
            InetSocketAddress remotePeer = new InetSocketAddress(nettyRpcNodeInfo.getHost(), nettyRpcNodeInfo.getIntPort());
            //连接结束
            logger.info("CONNECT TO REMOTE SERVER. REMOTE PEER = " + remotePeer);
            //同步阻塞建立连接
            Future future = executorService.submit(new NettyRpcServerConnectTask(nettyRpcNodeInfo, eventLoopGroup, nettyRpcClient));
            future.get();
            if (isConnected()) {
                return false;
            }
            if (logger.isInfoEnabled()) {
                logger.info("CONNECT SUCCESS.");
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }

    }

    public boolean isConnected() {
        if (channel == null) {
            return false;
        }
        return channel.isActive();
    }


    /**
     * 发送一条消息
     *
     * @return
     */
    public boolean sendRequestMessage(NettyRpcRequestMessage rpcRequestMessage) {
        if (!isConnected() && reConnectOn) {
            tryReConnect();
            if (!isConnected()) {
                return false;
            }
        }
        // 发送消息
        if (channel != null) {
            if (logger.isDebugEnabled()) {
                logger.debug("【SEND】" + rpcRequestMessage);
            }
            channel.writeAndFlush(rpcRequestMessage);
            return true;
        }
        return false;
    }

    public void tryReConnect() {
        statusLock.lock();
        try {
            if (!isConnected()) {
                try {
                    Future<?> future = executorService.submit(new ReConnect());
                    future.get();
                } catch (Exception e) {
                    logger.error("NETTY RPC CLIENT CONNECTION TRY RECONNECT IS ERROR");
                    logger.error(e);
                }
            }
        } catch (Exception e) {
        } finally {
            statusLock.unlock();
        }
    }

    /**
     * 重连线程内部类
     *
     * @author Fancy
     */
    private class ReConnect implements Runnable {

        @Override
        public void run() {
            try {
                open();
            } catch (Exception e) {
                if (logger.isDebugEnabled()) {
                    logger.error("RESTART CONNECTION ERROR.");
                }
            } finally {
                // 设置为允许重连
//                reConnect = false;
            }
        }
    }

    /**
     * 启动自动重连
     */
    public void setReconnectOn() {
        this.reConnectOn = true;
    }

    /**
     * 关闭自动重连
     */
    public void setReconnectOff() {
        this.reConnectOn = false;
    }

    public NioSocketChannel getChannel() {
        return channel;
    }

    public void setChannel(NioSocketChannel channel) {
        this.channel = channel;
    }

    public void close() {
        if (channel != null) {
            channel.close();
        }
        eventLoopGroup.shutdownGracefully();
    }

}
