content: package com.alibaba.otter.canal.admin.service.impl;
  
  import java.security.NoSuchAlgorithmException;
  import java.util.List;
  import java.util.Optional;
  import java.util.stream.Collectors;
  
  import org.apache.commons.lang.StringUtils;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.stereotype.Service;
  
  import com.alibaba.otter.canal.admin.common.exception.ServiceException;
  import com.alibaba.otter.canal.admin.model.CanalCluster;
  import com.alibaba.otter.canal.admin.model.CanalConfig;
  import com.alibaba.otter.canal.admin.model.CanalInstanceConfig;
  import com.alibaba.otter.canal.admin.model.NodeServer;
  import com.alibaba.otter.canal.admin.service.CanalClusterService;
  import com.alibaba.otter.canal.admin.service.NodeServerService;
  import com.alibaba.otter.canal.admin.service.PollingConfigService;
  import com.alibaba.otter.canal.protocol.SecurityUtil;
  import com.google.common.base.Joiner;
  
  @Service
  public class PollingConfigServiceImpl implements PollingConfigService {
  
      @Autowired
      NodeServerService   nodeServerService;
  
      @Autowired
      CanalClusterService canalClusterService;
  
      public boolean autoRegister(String ip, Integer adminPort, String cluster, String name) {
          NodeServer server = NodeServer.find.query().where().eq("ip", ip).eq("adminPort", adminPort).findOne();
          if (server == null) {
              server = new NodeServer();
              server.setName(Optional.ofNullable(name).orElse(ip));
              server.setIp(ip);
              server.setAdminPort(adminPort);
              server.setTcpPort(adminPort + 1);
              server.setMetricPort(adminPort + 2);
              if (StringUtils.isNotEmpty(cluster)) {
                  CanalCluster clusterConfig = CanalCluster.find.query().where().eq("name", cluster).findOne();
                  if (clusterConfig == null) {
                      throw new ServiceException("auto cluster : " + cluster + " is not found.");
                  }
  
                  server.setClusterId(clusterConfig.getId());
              }
              nodeServerService.save(server);
          }
  
          return true;
      }
  
      public CanalConfig getChangedConfig(String ip, Integer port, String md5) {
          NodeServer server = NodeServer.find.query().where().eq("ip", ip).eq("adminPort", port).findOne();
          if (server == null) {
              return null;
          }
          CanalConfig canalConfig;
          if (server.getClusterId() != null) { // 集群模式
              canalConfig = CanalConfig.find.query().where().eq("clusterId", server.getClusterId()).findOne();
          } else { // 单机模式
              canalConfig = CanalConfig.find.query().where().eq("serverId", server.getId()).findOne();
          }
          if (canalConfig == null) {
              throw new ServiceException("canal.properties config is empty");
          }
  
          if (!canalConfig.getContentMd5().equals(md5)) { // 内容发生变化
              return canalConfig;
          }
          return null;
      }
  
      public CanalInstanceConfig getInstancesConfig(String ip, Integer port, String md5) {
          NodeServer server = NodeServer.find.query().where().eq("ip", ip).eq("adminPort", port).findOne();
          if (server == null) {
              return null;
          }
          List<CanalInstanceConfig> canalInstanceConfigs;
          if (server.getClusterId() != null) { // 集群模式
              canalInstanceConfigs = CanalInstanceConfig.find.query()
                  .where()
                  .eq("status", "1")
                  .eq("clusterId", server.getClusterId())
                  .findList(); // 取属于该集群的所有instance config
          } else { // 单机模式
              canalInstanceConfigs = CanalInstanceConfig.find.query()
                  .where()
                  .eq("status", "1")
                  .eq("serverId", server.getId())
                  .findList();
          }
  
          CanalInstanceConfig canalInstanceConfig = new CanalInstanceConfig();
          List<String> instances = canalInstanceConfigs.stream()
              .map(CanalInstanceConfig::getName)
              .collect(Collectors.toList());
          String data = Joiner.on(',').join(instances);
          canalInstanceConfig.setContent(data);
          if (!StringUtils.isEmpty(md5)) {
              try {
                  String newMd5 = SecurityUtil.md5String(canalInstanceConfig.getContent());
                  if (StringUtils.equals(md5, newMd5)) {
                      canalInstanceConfig.setContent(null);
                  }
              } catch (NoSuchAlgorithmException e) {
                  // ignore
              }
          }
          return canalInstanceConfig;
      }
  
      public CanalInstanceConfig getInstanceConfig(String destination, String md5) {
          CanalInstanceConfig instanceConfig = CanalInstanceConfig.find.query().where().eq("name", destination).findOne();
          if (instanceConfig == null) {
              return null;
          }
          if (StringUtils.isEmpty(md5)) {
              return instanceConfig;
          } else {
              try {
                  String newMd5 = SecurityUtil.md5String(instanceConfig.getContent());
                  if (StringUtils.equals(md5, newMd5)) {
                      instanceConfig.setContent(null);
                  }
              } catch (NoSuchAlgorithmException e) {
                  // ignore
              }
  
              return instanceConfig;
          }
      }
  }
  