package com.twjitm.core.common.zookeeper;

import com.twjitm.core.common.config.global.NettyGameServiceConfig;
import com.twjitm.core.common.config.global.NettyGameServiceConfigService;
import com.twjitm.core.common.config.global.ZookeeperConfig;
import com.twjitm.core.common.config.rpc.RpcServerConfig;
import com.twjitm.core.common.service.IService;
import com.twjitm.core.common.service.rpc.server.NettySdRpcServiceProvider;
import com.twjitm.core.spring.SpringServiceManager;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

/**
 * @author twjtim - [Created on 2018-08-22 11:16]
 * @jdk java version "1.8.0_77"
 */
public class NettyZookeeperRpcServiceRegistryService implements IService {
    private static final Logger logger = LoggerFactory.getLogger(NettyZookeeperRpcServiceRegistryService.class);
    private CountDownLatch countDownLatch = new CountDownLatch(1);
    private ZooKeeper zooKeeper;

    @Override
    public String getId() {
        return NettyZookeeperRpcServiceRegistryService.class.getSimpleName();
    }

    @Override
    public void startup() throws Exception {
        logger.info("STARTUP ZOOKEEPER SERVER BEGINNING");
        NettyGameServiceConfigService gameServiceConfigService = SpringServiceManager.getSpringLoadService().getNettyGameServiceConfigService();
        boolean isOpen = gameServiceConfigService.getNettyGameServiceConfig().isZookeeperOpen();
        if (isOpen) {
            registryZookeeper();
            registryZookeeperNode();
        }


    }

    /**
     * 在启动服务器的时候，需要将服务器信息注册到zookeeper，提供给其他服务器使用，当服务器不可达时候，zookeeper会将
     * 服务器信息清楚，达到容灾策略。
     */
    private void registryZookeeperNode() {
        NettyGameServiceConfigService gameConfig = SpringServiceManager.getSpringLoadService().getNettyGameServiceConfigService();
        RpcServerConfig rpcConfig = gameConfig.getRpcServerConfig();
        NettyGameServiceConfig gameServiceConfig = gameConfig.getNettyGameServiceConfig();
        NettySdRpcServiceProvider nettySdRpcServiceProvider = rpcConfig.getProvider();
        boolean worldOpen = nettySdRpcServiceProvider.isWorldOpen();
        if (worldOpen) {
            NettyZookeeperNodeInfo zookeeperNodeInfo = new NettyZookeeperNodeInfo(NettyZookeeperNodeNettyGameTypeEnum.WORLD,
                    gameServiceConfig.getServerId(),
                    gameServiceConfig.getServerHost(),
                    gameServiceConfig.getServerPort());
            registry(zookeeperNodeInfo);
            logger.info("注册world节点到zookeeper");
        }
        if (nettySdRpcServiceProvider.isGameOpen()) {
            NettyZookeeperNodeInfo zookeeperNodeInfo = new NettyZookeeperNodeInfo(NettyZookeeperNodeNettyGameTypeEnum.GAME,
                    gameServiceConfig.getServerId(),
                    gameServiceConfig.getServerHost(),
                    gameServiceConfig.getServerPort());
            registry(zookeeperNodeInfo);
            logger.info("注册game节点到zookeeper");
        }

        if (nettySdRpcServiceProvider.isDbOpen()) {
            NettyZookeeperNodeInfo zookeeperNodeInfo = new NettyZookeeperNodeInfo(NettyZookeeperNodeNettyGameTypeEnum.DB,
                    gameServiceConfig.getServerId(),
                    gameServiceConfig.getServerHost(),
                    gameServiceConfig.getServerPort());
            registry(zookeeperNodeInfo);
            logger.info("注册db节点到zookeeper");
        }

    }

    private void registry(NettyZookeeperNodeInfo zookeeperNodeInfo) {
        String rootPath = zookeeperNodeInfo.getNettyZookeeperNodeNettyGameTypeEnum().getRootPath();
        String nodePath = zookeeperNodeInfo.getZookeeperNodePath();
        String nodeData = null;
        try {
            nodeData = zookeeperNodeInfo.serialize();
        } catch (IOException e) {
            e.printStackTrace();
        }
        if (!StringUtils.isEmpty(rootPath)) {
            if (zooKeeper != null) {
                addRootNode(zooKeeper, rootPath);
                try {
                    if (zooKeeper.exists(nodePath, false) != null) {
                        deleteNode(zooKeeper, nodePath);
                        logger.info("DELETE ZOOKEEPER NODE ", nodeData);
                    }
                } catch (KeeperException e) {
                } catch (InterruptedException e) {
                }
                createNode(zooKeeper, nodePath, nodeData);
                logger.info("CREATE ZOOKEEPER NODE " + nodeData);
            }
        }
    }

    private void registryZookeeper() {
        if (zooKeeper == null) {
            zooKeeper = connectZookeeperServer();
        }
    }

    /**
     * 链接到zookeeper 服务器
     *
     * @return
     */
    private ZooKeeper connectZookeeperServer() {
        ZooKeeper zk = null;
        try {
            NettyGameServiceConfigService gameServerConfigService = SpringServiceManager.getSpringLoadService().getNettyGameServiceConfigService();
            ZookeeperConfig zooKeeperConfig = gameServerConfigService.getZookeeperConfig();
            String hostAndPort = zooKeeperConfig.getServerHost() + ":" + zooKeeperConfig.getServerPort();
            zk = new ZooKeeper(hostAndPort, zooKeeperConfig.getTimeOut(),
                    event -> countDownLatch.countDown());
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
        return zk;
    }

    private void addRootNode(ZooKeeper zk, String rootPath) {
        try {
            Stat s = zk.exists(rootPath, false);
            if (s == null) {
                createRootNode(rootPath, new byte[0]);
                logger.info("CREATE ZOOKEEPER ROOT NODE " + rootPath);
            }
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }

    }

    private void createNode(ZooKeeper zk, String nodePath, String nodeData) {
        try {
            byte[] bytes = nodeData.getBytes();
            String path = create(nodePath, bytes);
            logger.debug("CREATE ZOOKEEPER NODE ({} => {})", path, bytes);
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
    }

    /**
     * <b>function:</b>创建持久态的znode,比支持多层创建.比如在创建/parent/child的情况下,无/parent.无法通过
     *
     * @param path
     * @param data
     * @throws KeeperException
     * @throws InterruptedException
     */
    public String createRootNode(String path, byte[] data) throws Exception {
        /**
         * 此处采用的是CreateMode是PERSISTENT  表示The znode will not be automatically deleted upon client's disconnect.
         * EPHEMERAL 表示The znode will be deleted upon the client's disconnect.
         */
        return this.zooKeeper.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }


    /**
     * <b>function:</b>创建持久态的znode,比支持多层创建.比如在创建/parent/child的情况下,无/parent.无法通过
     *
     * @param path
     * @param data
     */
    public String create(String path, byte[] data) throws Exception {
        /**
         * 此处采用的是CreateMode是PERSISTENT  表示The znode will not be automatically deleted upon client's disconnect.
         * EPHEMERAL 表示The znode will be deleted upon the client's disconnect.
         */
        return this.zooKeeper.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    }


    public void deleteNode(ZooKeeper zk, String nodePath) {
        try {
            zk.delete(nodePath, -1);
            logger.debug("delete zookeeper node path c ({} => {})", nodePath);
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
    }


    @Override
    public void shutdown() throws Exception {
        if (zooKeeper != null) {
            zooKeeper.close();
        }

    }
}
