content: /*
   * Copyright 1999-2018 Alibaba Group Holding Ltd.
   *
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   *      http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package com.alibaba.nacos.client.naming.beat;
  
  import com.alibaba.nacos.api.common.Constants;
  import com.alibaba.nacos.client.monitor.MetricsMonitor;
  import com.alibaba.nacos.client.naming.net.NamingProxy;
  import com.alibaba.nacos.client.naming.utils.UtilAndComs;
  
  import java.util.Map;
  import java.util.concurrent.*;
  
  import static com.alibaba.nacos.client.utils.LogUtils.NAMING_LOGGER;
  
  /**
   * @author harold
   */
  public class BeatReactor {
  
      private ScheduledExecutorService executorService;
  
      private NamingProxy serverProxy;
  
      public final Map<String, BeatInfo> dom2Beat = new ConcurrentHashMap<String, BeatInfo>();
  
      public BeatReactor(NamingProxy serverProxy) {
          this(serverProxy, UtilAndComs.DEFAULT_CLIENT_BEAT_THREAD_COUNT);
      }
  
      public BeatReactor(NamingProxy serverProxy, int threadCount) {
          this.serverProxy = serverProxy;
  
          executorService = new ScheduledThreadPoolExecutor(threadCount, new ThreadFactory() {
              @Override
              public Thread newThread(Runnable r) {
                  Thread thread = new Thread(r);
                  thread.setDaemon(true);
                  thread.setName("com.alibaba.nacos.naming.beat.sender");
                  return thread;
              }
          });
      }
  
      public void addBeatInfo(String serviceName, BeatInfo beatInfo) {
          NAMING_LOGGER.info("[BEAT] adding beat: {} to beat map.", beatInfo);
          dom2Beat.put(buildKey(serviceName, beatInfo.getIp(), beatInfo.getPort()), beatInfo);
          executorService.schedule(new BeatTask(beatInfo), beatInfo.getPeriod(), TimeUnit.MILLISECONDS);
          MetricsMonitor.getDom2BeatSizeMonitor().set(dom2Beat.size());
      }
  
      public void removeBeatInfo(String serviceName, String ip, int port) {
          NAMING_LOGGER.info("[BEAT] removing beat: {}:{}:{} from beat map.", serviceName, ip, port);
          BeatInfo beatInfo = dom2Beat.remove(buildKey(serviceName, ip, port));
          if (beatInfo == null) {
              return;
          }
          beatInfo.setStopped(true);
          MetricsMonitor.getDom2BeatSizeMonitor().set(dom2Beat.size());
      }
  
      private String buildKey(String serviceName, String ip, int port) {
          return serviceName + Constants.NAMING_INSTANCE_ID_SPLITTER
              + ip + Constants.NAMING_INSTANCE_ID_SPLITTER + port;
      }
  
      class BeatTask implements Runnable {
  
          BeatInfo beatInfo;
  
          public BeatTask(BeatInfo beatInfo) {
              this.beatInfo = beatInfo;
          }
  
          @Override
          public void run() {
              if (beatInfo.isStopped()) {
                  return;
              }
              long result = serverProxy.sendBeat(beatInfo);
              long nextTime = result > 0 ? result : beatInfo.getPeriod();
              executorService.schedule(new BeatTask(beatInfo), nextTime, TimeUnit.MILLISECONDS);
          }
      }
  }
  