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.service.IService;
import com.twjitm.core.common.service.rpc.service.NettyRpcClientConnectService;
import com.twjitm.core.spring.SpringServiceManager;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.ACLProvider;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 采用Apache Curator 来发现zookeeper服务。
 *
 * @author twjitm - [Created on 2018-08-22 15:57]
 * @company https://github.com/twjitm/
 * @jdk java version "1.8.0_77"
 * zookeeper 服务发现
 */

public class NettyZookeeperRpcServiceDiscoveryService implements IService {
    private Logger logger = LoggerFactory.getLogger(NettyZookeeperRpcServiceDiscoveryService.class);
    private volatile ConcurrentHashMap<NettyZookeeperNodeNettyGameTypeEnum, List<NettyZookeeperNodeInfo>> nettyZookeeperNodeMap;
    /**
     * zookeeper连接客户端   https://blog.csdn.net/dc_726/article/details/46475633
     */
    private CuratorFramework client;

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

    public List<NettyZookeeperNodeInfo> getNodeList(final NettyZookeeperNodeNettyGameTypeEnum zooKeeperNodeBoEnum) {
        return nettyZookeeperNodeMap.get(zooKeeperNodeBoEnum);
    }

    @Override
    public void startup() throws Exception {
        nettyZookeeperNodeMap = new ConcurrentHashMap<>();
        NettyGameServiceConfigService gameConfig = SpringServiceManager.getSpringLoadService().getNettyGameServiceConfigService();
        NettyGameServiceConfig gameServiceConfig = gameConfig.getNettyGameServiceConfig();
        boolean openZookeeper = gameServiceConfig.isZookeeperOpen();
        //创建一个连接
        if (openZookeeper) {
            try {
                client = createCuratorClient();
            } catch (Exception e) {
                logger.error("ERROR IN NettyZookeeperRpcServiceDiscoveryService startup()", e);
            }

            NettyZookeeperNodeNettyGameTypeEnum[] nodeNettyGameTypeEnum = NettyZookeeperNodeNettyGameTypeEnum.values();
            for (NettyZookeeperNodeNettyGameTypeEnum temp : nodeNettyGameTypeEnum) {
                discoveryService(temp);
            }

        }

    }


    private void discoveryService(NettyZookeeperNodeNettyGameTypeEnum nettyZookeeperNodeNettyGameTypeEnum) {
        if (client == null) {
            client = createCuratorClient();
        }
        try {
            setCuratorListener(client, nettyZookeeperNodeNettyGameTypeEnum);
        } catch (Exception e) {
            if (logger.isDebugEnabled()) {
                logger.debug("CURATORFRAMEWORK LISTENING EXCEPTION:" + e.getMessage());
            }
        }
    }

    /**
     * 设置一个监听器，负责监听zookeeper信息变化回调
     * Apache Curator
     *
     * @param client
     * @param nettyZookeeperNodeNettyGameTypeEnum
     */
    private void setCuratorListener(CuratorFramework client, NettyZookeeperNodeNettyGameTypeEnum nettyZookeeperNodeNettyGameTypeEnum) throws Exception {

        List<String> childRenList = client.getChildren().forPath(nettyZookeeperNodeNettyGameTypeEnum.getRootPath());
        List<NettyZookeeperNodeInfo> tempNodeList = new ArrayList<>();
        for (String node : childRenList) {
            NettyZookeeperNodeInfo zooKeeperNodeInfo = new NettyZookeeperNodeInfo();
            byte[] bytes = client.getData().forPath(nettyZookeeperNodeNettyGameTypeEnum.getRootPath() + "/" + node);
            if (bytes != null) {
                zooKeeperNodeInfo.deserialize(new String(bytes));
                tempNodeList.add(zooKeeperNodeInfo);
            }
        }

        if (logger.isDebugEnabled()) {
            logger.debug("NODE DATA: {}", tempNodeList);
        }

        nettyZookeeperNodeMap.put(nettyZookeeperNodeNettyGameTypeEnum, tempNodeList);
        if (logger.isDebugEnabled()) {
            logger.debug("SERVICE DISCOVERY TRIGGERED UPDATING CONNECTED SERVER NODE.");
        }

        NettyRpcClientConnectService rpcClientConnectService = SpringServiceManager.getSpringLoadService().getNettyRpcClientConnectService();
        rpcClientConnectService.notifyConnect(nettyZookeeperNodeNettyGameTypeEnum, nettyZookeeperNodeMap.get(nettyZookeeperNodeNettyGameTypeEnum));

        TreeCache cache = new TreeCache(client, nettyZookeeperNodeNettyGameTypeEnum.getRootPath());
        cache.getListenable().addListener((client1, event) -> {
            ChildData data = event.getData();
            if (data != null) {
                switch (event.getType()) {
                    case NODE_ADDED:
                        if (logger.isDebugEnabled()) {
                            logger.debug("NODE_ADDED : " + data.getPath() + "  DATA:" + new String(data.getData()));
                        }
                        break;
                    case NODE_REMOVED:
                        if (logger.isDebugEnabled()) {
                            logger.debug("NODE_REMOVED : " + data.getPath() + "  DATA:" + new String(data.getData()));
                        }
                        break;
                    case NODE_UPDATED:
                        if (logger.isDebugEnabled()) {
                            logger.debug("NODE_UPDATED : " + data.getPath() + "  DATA:" + new String(data.getData()));
                        }
                        break;
                    default:
                        break;
                }
            } else {
                switch (event.getType()) {
                    case CONNECTION_SUSPENDED:
                        if (logger.isDebugEnabled()) {
                            logger.debug("DATA IS NULL : " + "CONNECTION_SUSPENDED");
                        }
                        break;
                    case CONNECTION_RECONNECTED:
                        if (logger.isDebugEnabled()) {
                            logger.debug("DATA IS NULL : " + "CONNECTION_RECONNECTED");
                        }
                        break;
                    case CONNECTION_LOST:
                        if (logger.isDebugEnabled()) {
                            logger.debug("DATA IS NULL : " + "CONNECTION_LOST");
                        }
                        break;
                    default:
                        break;
                }
            }
        });
        // 开始监听
        cache.start();
    }


    private CuratorFramework createCuratorClient() {
        /**
         * ACL 提供者
         */
        ACLProvider aclProvider = new ACLProvider() {
            private List<ACL> acl;

            @Override
            public List<ACL> getDefaultAcl() {
                if (acl == null) {
                    ArrayList<ACL> acl = ZooDefs.Ids.CREATOR_ALL_ACL;
                    acl.clear();
                    acl.add(new ACL(ZooDefs.Perms.ALL, new Id("auth", "admin:admin")));
                    this.acl = acl;
                }
                return acl;
            }

            @Override
            public List<ACL> getAclForPath(String path) {
                return acl;
            }
        };


        int connectionTimeoutMs = SpringServiceManager.getSpringLoadService().getNettyGameServiceConfigService().getZookeeperConfig().getTimeOut();
        NettyGameServiceConfigService gameServerConfigService = SpringServiceManager.getSpringLoadService().getNettyGameServiceConfigService();
        ZookeeperConfig zooKeeperConfig = gameServerConfigService.getZookeeperConfig();
        String registryAddress = zooKeeperConfig.getRegistryAddress();

        CuratorFramework client = CuratorFrameworkFactory.builder()
                .aclProvider(aclProvider)
                .connectionTimeoutMs(connectionTimeoutMs)
                .connectString(registryAddress)
                .retryPolicy(
                        new RetryNTimes(Integer.MAX_VALUE, zooKeeperConfig.getElapsedTimeMs())
                ).build();//绑定
        //启动
        client.start();
        return client;
    }

    @Override
    public void shutdown() throws Exception {
        if (client != null) {
            try {
                client.close();
            } catch (Exception e) {
                logger.error(e.toString(), e);
            }
        }
    }
}
