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

import com.twjitm.core.common.config.global.NettyGameServiceConfig;
import com.twjitm.core.common.config.global.NettyGameServiceConfigService;
import com.twjitm.core.common.service.rpc.network.NettyRpcClient;
import com.twjitm.core.common.service.rpc.server.NettyRpcNodeInfo;
import com.twjitm.core.common.service.rpc.server.NettySdServer;
import com.twjitm.core.common.zookeeper.NettyZookeeperNodeInfo;
import com.twjitm.core.spring.SpringServiceManager;
import com.twjitm.core.utils.logs.LoggerUtils;
import org.apache.log4j.Logger;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author twjitm - [Created on 2018-08-20 10:57]
 * @company https://github.com/twjitm/
 * @jdk java version "1.8.0_77"
 * rpc 抽象连接管理器
 */
public abstract class AbstractNettyRpcConnectManager {
    private Logger logger = LoggerUtils.getLogger(AbstractNettyRpcConnectManager.class);

    private ThreadPoolExecutor threadPoolExecutor;

    private ReentrantLock lock = new ReentrantLock();
    private Map<Integer, NettyRpcClient> serverNodes = new HashMap<>();

    private AtomicInteger roundRobin = new AtomicInteger();


    public void initManager() {
        NettyGameServiceConfigService config = SpringServiceManager.getSpringLoadService().getNettyGameServiceConfigService();
        NettyGameServiceConfig serviceConfig = config.getNettyGameServiceConfig();
        serviceConfig.getAsyncThreadPoolMaxSize();
        threadPoolExecutor = new ThreadPoolExecutor(serviceConfig.getRpcConnectThreadSize(),
                serviceConfig.getRpcConnectThreadSize(),
                600L, TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(65536));

    }

    /**
     * 本服务器自己配置在本地的rpc
     *
     * @param allServerAddress
     * @throws InterruptedException
     */
    public void initServers(List<NettySdServer> allServerAddress) throws InterruptedException {
        lock.lock();
        try {
            if (allServerAddress != null) {
                for (NettySdServer sdServer : allServerAddress) {
                    if (serverNodes.containsKey(sdServer.getServerId())) {
                        continue;
                    }
                    NettyRpcNodeInfo rpcNodeInfo = new NettyRpcNodeInfo();
                    rpcNodeInfo.setServerId(String.valueOf(sdServer.getServerId()));
                    rpcNodeInfo.setHost(sdServer.getIp());
                    rpcNodeInfo.setPort(String.valueOf(sdServer.getRpcPort()));
                    NettyRpcClient rpcClient = new NettyRpcClient(rpcNodeInfo, threadPoolExecutor);
                    serverNodes.put(sdServer.getServerId(), rpcClient);
                }
            }
        } catch (Exception e) {
            logger.error(e);
        } finally {
            lock.unlock();
        }

    }

    /**
     * 选择远程rpc
     *
     * @param serverId
     * @return
     */
    public NettyRpcClient getNettyRpcClientByServerId(int serverId) {
        if (serverId == 0) {
            List<NettyRpcClient> handlers = new ArrayList(this.serverNodes.values());
            int size = handlers.size();
            int index = (roundRobin.getAndAdd(1) + size) % size;
            return handlers.get(index);
        } else {
            try {
                NettyRpcClient rpcClient = this.serverNodes.get(serverId);
                return rpcClient;
            } catch (Exception e) {
                logger.error("WAITING FOR AVAILABLE NODE IS INTERRUPTED! ");
                logger.error(e);
                throw new RuntimeException("CAN'T CONNECT ANY SERVERS!", e);
            }
        }
    }

    public void stop() {
        for (NettyRpcClient rpcClient : serverNodes.values()) {
            rpcClient.close();
        }
        if (threadPoolExecutor != null) {
            threadPoolExecutor.shutdown();
        }
    }


    /**
     * 通过zookeeper发现的rpc
     *
     * @param nettyZookeeperNodeInfoList
     */
    public void initNettyZookeeperRpcServers(List<NettyZookeeperNodeInfo> nettyZookeeperNodeInfoList) {
        //增加同步，当前
        synchronized (this) {
            if (nettyZookeeperNodeInfoList != null) {
                //不能把自己添加到这里面，应为自己已经添加进去了，应该添加别的的服务器信息
                for (NettyZookeeperNodeInfo zooKeeperNodeInfo : nettyZookeeperNodeInfoList) {
                    if (serverNodes.containsKey(zooKeeperNodeInfo.getServerId())) {
                        continue;
                    }
                    NettyRpcNodeInfo rpcNodeInfo = new NettyRpcNodeInfo();
                    rpcNodeInfo.setServerId(zooKeeperNodeInfo.getServerId());
                    rpcNodeInfo.setHost(zooKeeperNodeInfo.getHost());
                    rpcNodeInfo.setPort(zooKeeperNodeInfo.getPort());
                    NettyRpcClient rpcClient = new NettyRpcClient(rpcNodeInfo, threadPoolExecutor);
                    serverNodes.put(Integer.parseInt(zooKeeperNodeInfo.getServerId()), rpcClient);
                }
            }
        }
    }
}
