// Path: trpc-core/src/main/java/com/github/sofn/trpc/core/AbstractMonitor.java
// public abstract class AbstractMonitor {
// 
//     /**
//      * 返回remoteKey下所有节点数据，并监控变化
//      *
//      * @param monitor 消息listener
//      * @return 返回现在的数据
//      */
//     public abstract List<RegistryConfig> monitorRemoteKey(RegistryConfigListener monitor);
// 
// }
// 
// Path: trpc-core/src/main/java/com/github/sofn/trpc/core/config/RegistryConfig.java
// @Setter
// @Getter
// @Builder
// @EqualsAndHashCode(of = {"serverInfo", "appKey"}, callSuper = false)
// public class RegistryConfig extends AbstractConfig {
//     @Singular
//     private List<ServiceConfig> servers = Collections.emptyList();
//     private ThriftServerInfo serverInfo;
//     private String appKey;
//     private String registry;
//     private String hostName;
//     private int weight; //权重
// 
//     public RegistryConfig() {
//         super("registry");
//     }
// 
//     public RegistryConfig(List<ServiceConfig> servers, ThriftServerInfo serverInfo, String appKey, String registry, String hostName, int weight) {
//         this();
//         this.servers = servers;
//         this.serverInfo = serverInfo;
//         this.appKey = appKey;
//         this.registry = registry;
//         this.hostName = hostName;
//         this.weight = weight;
//     }
// 
//     public String toJsonString() {
//         this.servers.stream().forEach(AbstractConfig::fillId);
//         this.fillId();
//         return JsonUtils.toJson(this);
//     }
// 
//     public static RegistryConfig parse(String json) {
//         return JsonUtils.fromJson(json, RegistryConfig.class);
//     }
// 
// }
// 
// Path: trpc-core/src/main/java/com/github/sofn/trpc/core/config/ThriftServerInfo.java
// @Data
// @AllArgsConstructor
// @EqualsAndHashCode
// public class ThriftServerInfo {
//     private String ip;
//     private int port;
// 
//     public ThriftServerInfo(String ipPort) {
//         checkNotNull(ipPort);
//         String[] splits = ipPort.split(":");
//         checkArgument(splits.length == 2, "ipPort format error");
//         this.ip = splits[0];
//         this.port = NumberUtils.toInt(splits[1]);
//     }
// 
//     @Override
//     public String toString() {
//         return ip + ":" + port;
//     }
// 
//     public static ThriftServerInfo parse(String info) {
//         String[] splits = StringUtils.split(info, ":");
//         if (splits == null || splits.length != 2) {
//             return null;
//         }
//         return new ThriftServerInfo(splits[0], NumberUtils.toInt(splits[1]));
//     }
// }
// 
// Path: trpc-core/src/main/java/com/github/sofn/trpc/core/monitor/RegistryConfigListener.java
// @Getter
// @AllArgsConstructor
// public abstract class RegistryConfigListener {
//     private String remoteKey;
// 
//     public abstract void addServer(RegistryConfig config);
// 
//     public abstract void removeServer(ThriftServerInfo serverInfo);
// 
//     public abstract void updateServer(RegistryConfig config);
// }


import com.github.sofn.trpc.core.AbstractMonitor;
import com.github.sofn.trpc.core.config.RegistryConfig;
import com.github.sofn.trpc.core.config.ThriftServerInfo;
import com.github.sofn.trpc.core.monitor.RegistryConfigListener;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.retry.ExponentialBackoffRetry;
import java.util.List;
import java.util.stream.Collectors;
    private void initMonitor() {
        client = CuratorFrameworkFactory.builder()
                .connectString(connectString)
                .sessionTimeoutMs(sessionTimeout)
                .connectionTimeoutMs(connectionTimeout)
                .canBeReadOnly(false)
                .retryPolicy(new ExponentialBackoffRetry(retrySleepTime, maxRetries))
                .namespace(ZkConstant.NAMESPACE)
                .build();
        client.start();
        client.getConnectionStateListenable().addListener((client1, newState)
                -> log.info("zookeeper client " + connectString + " change status: " + newState.name()));
        log.info("init zookeeper connect: " + connectString);
    }

    public CuratorFramework getClient() {
        if (this.client == null) {
            initMonitor();
        }
        return client;
    }

    @Override
    public List<RegistryConfig> monitorRemoteKey(RegistryConfigListener listener) {
        try {
            final PathChildrenCache childrenCache = new PathChildrenCache(getClient(), ZkConstant.SERVICES_DIR + listener.getRemoteKey(), true);
            childrenCache.getListenable().addListener(
                    (client1, event) -> {
                        switch (event.getType()) {
                            case CHILD_ADDED:
// Path: trpc-core/src/main/java/com/github/sofn/trpc/core/AbstractMonitor.java
// public abstract class AbstractMonitor {
// 
//     /**
//      * 返回remoteKey下所有节点数据，并监控变化
//      *
//      * @param monitor 消息listener
//      * @return 返回现在的数据
//      */
//     public abstract List<RegistryConfig> monitorRemoteKey(RegistryConfigListener monitor);
// 
// }
// 
// Path: trpc-core/src/main/java/com/github/sofn/trpc/core/config/RegistryConfig.java
// @Setter
// @Getter
// @Builder
// @EqualsAndHashCode(of = {"serverInfo", "appKey"}, callSuper = false)
// public class RegistryConfig extends AbstractConfig {
//     @Singular
//     private List<ServiceConfig> servers = Collections.emptyList();
//     private ThriftServerInfo serverInfo;
//     private String appKey;
//     private String registry;
//     private String hostName;
//     private int weight; //权重
// 
//     public RegistryConfig() {
//         super("registry");
//     }
// 
//     public RegistryConfig(List<ServiceConfig> servers, ThriftServerInfo serverInfo, String appKey, String registry, String hostName, int weight) {
//         this();
//         this.servers = servers;
//         this.serverInfo = serverInfo;
//         this.appKey = appKey;
//         this.registry = registry;
//         this.hostName = hostName;
//         this.weight = weight;
//     }
// 
//     public String toJsonString() {
//         this.servers.stream().forEach(AbstractConfig::fillId);
//         this.fillId();
//         return JsonUtils.toJson(this);
//     }
// 
//     public static RegistryConfig parse(String json) {
//         return JsonUtils.fromJson(json, RegistryConfig.class);
//     }
// 
// }
// 
// Path: trpc-core/src/main/java/com/github/sofn/trpc/core/config/ThriftServerInfo.java
// @Data
// @AllArgsConstructor
// @EqualsAndHashCode
// public class ThriftServerInfo {
//     private String ip;
//     private int port;
// 
//     public ThriftServerInfo(String ipPort) {
//         checkNotNull(ipPort);
//         String[] splits = ipPort.split(":");
//         checkArgument(splits.length == 2, "ipPort format error");
//         this.ip = splits[0];
//         this.port = NumberUtils.toInt(splits[1]);
//     }
// 
//     @Override
//     public String toString() {
//         return ip + ":" + port;
//     }
// 
//     public static ThriftServerInfo parse(String info) {
//         String[] splits = StringUtils.split(info, ":");
//         if (splits == null || splits.length != 2) {
//             return null;
//         }
//         return new ThriftServerInfo(splits[0], NumberUtils.toInt(splits[1]));
//     }
// }
// 
// Path: trpc-core/src/main/java/com/github/sofn/trpc/core/monitor/RegistryConfigListener.java
// @Getter
// @AllArgsConstructor
// public abstract class RegistryConfigListener {
//     private String remoteKey;
// 
//     public abstract void addServer(RegistryConfig config);
// 
//     public abstract void removeServer(ThriftServerInfo serverInfo);
// 
//     public abstract void updateServer(RegistryConfig config);
// }


// Path: trpc-registry/trpc-registry-zookeeper/src/main/java/com/github/sofn/trpc/registry/zk/ZkMonitor.java
import com.github.sofn.trpc.core.AbstractMonitor;
import com.github.sofn.trpc.core.config.RegistryConfig;
import com.github.sofn.trpc.core.config.ThriftServerInfo;
import com.github.sofn.trpc.core.monitor.RegistryConfigListener;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.retry.ExponentialBackoffRetry;
import java.util.List;
import java.util.stream.Collectors;

    private void initMonitor() {
        client = CuratorFrameworkFactory.builder()
                .connectString(connectString)
                .sessionTimeoutMs(sessionTimeout)
                .connectionTimeoutMs(connectionTimeout)
                .canBeReadOnly(false)
                .retryPolicy(new ExponentialBackoffRetry(retrySleepTime, maxRetries))
                .namespace(ZkConstant.NAMESPACE)
                .build();
        client.start();
        client.getConnectionStateListenable().addListener((client1, newState)
                -> log.info("zookeeper client " + connectString + " change status: " + newState.name()));
        log.info("init zookeeper connect: " + connectString);
    }

    public CuratorFramework getClient() {
        if (this.client == null) {
            initMonitor();
        }
        return client;
    }

    @Override
    public List<RegistryConfig> monitorRemoteKey(RegistryConfigListener listener) {
        try {
            final PathChildrenCache childrenCache = new PathChildrenCache(getClient(), ZkConstant.SERVICES_DIR + listener.getRemoteKey(), true);
            childrenCache.getListenable().addListener(
                    (client1, event) -> {
                        switch (event.getType()) {
                            case CHILD_ADDED:
                                listener.addServer(RegistryConfig.parse(new String(event.getData().getData())));