content: package com.alibaba.otter.canal.common;
  
  import java.util.concurrent.CountDownLatch;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  
  import org.apache.commons.lang.math.RandomUtils;
  import org.junit.After;
  import org.junit.Assert;
  import org.junit.Before;
  import org.junit.Ignore;
  import org.junit.Test;
  
  import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
  import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils;
  import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningData;
  import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningListener;
  import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningMonitor;
  
  @Ignore
  public class ServerRunningTest extends AbstractZkTest {
  
      private ZkClientx zkclientx = new ZkClientx(cluster1 + ";" + cluster2);
  
      @Before
      public void setUp() {
          String path = ZookeeperPathUtils.getDestinationPath(destination);
          zkclientx.deleteRecursive(path);
  
          zkclientx.createPersistent(ZookeeperPathUtils.getDestinationPath(destination), true);
      }
  
      @After
      public void tearDown() {
          String path = ZookeeperPathUtils.getDestinationPath(destination);
          zkclientx.deleteRecursive(path);
      }
  
      @Test
      public void testOneServer() {
          final CountDownLatch countLatch = new CountDownLatch(2);
          ServerRunningMonitor runningMonitor = buildServerRunning(countLatch, "127.0.0.1", 2088);
          runningMonitor.start();
          sleep(2000L);
          runningMonitor.stop();
          sleep(2000L);
  
          if (countLatch.getCount() != 0) {
              Assert.fail();
          }
      }
  
      @Test
      public void testMultiServer() {
          final CountDownLatch countLatch = new CountDownLatch(30);
          final ServerRunningMonitor runningMonitor1 = buildServerRunning(countLatch, "127.0.0.1", 2088);
          final ServerRunningMonitor runningMonitor2 = buildServerRunning(countLatch, "127.0.0.1", 2089);
          final ServerRunningMonitor runningMonitor3 = buildServerRunning(countLatch, "127.0.0.1", 2090);
          final ExecutorService executor = Executors.newFixedThreadPool(3);
          executor.submit(() -> {
              for (int i = 0; i < 10; i++) {
                  if (!runningMonitor1.isStart()) {
                      runningMonitor1.start();
                  }
                  sleep(2000L + RandomUtils.nextInt(500));
                  if (runningMonitor1.check()) {
                      runningMonitor1.stop();
                  }
                  sleep(2000L + RandomUtils.nextInt(500));
              }
          });
  
          executor.submit(() -> {
              for (int i = 0; i < 10; i++) {
                  if (!runningMonitor2.isStart()) {
                      runningMonitor2.start();
                  }
                  sleep(2000L + RandomUtils.nextInt(500));
                  if (runningMonitor2.check()) {
                      runningMonitor2.stop();
                  }
                  sleep(2000L + RandomUtils.nextInt(500));
              }
          });
  
          executor.submit(() -> {
              for (int i = 0; i < 10; i++) {
                  if (!runningMonitor3.isStart()) {
                      runningMonitor3.start();
                  }
                  sleep(2000L + RandomUtils.nextInt(500));
                  if (runningMonitor3.check()) {
                      runningMonitor3.stop();
                  }
                  sleep(2000L + RandomUtils.nextInt(500));
              }
          });
  
          sleep(30000L);
      }
  
      private ServerRunningMonitor buildServerRunning(final CountDownLatch countLatch, final String ip, final int port) {
          ServerRunningData serverData = new ServerRunningData(ip + ":" + port);
          ServerRunningMonitor runningMonitor = new ServerRunningMonitor(serverData);
          runningMonitor.setDestination(destination);
          runningMonitor.setListener(new ServerRunningListener() {
  
              public void processActiveEnter() {
                  System.out.println(String.format("ip:%s:%s has start", ip, port));
                  countLatch.countDown();
              }
  
              public void processActiveExit() {
                  System.out.println(String.format("ip:%s:%s has stop", ip, port));
                  countLatch.countDown();
              }
  
              public void processStart() {
                  System.out.println(String.format("ip:%s:%s processStart", ip, port));
              }
  
              public void processStop() {
                  System.out.println(String.format("ip:%s:%s processStop", ip, port));
              }
  
          });
  
          runningMonitor.setZkClient(zkclientx);
          runningMonitor.setDelayTime(1);
          return runningMonitor;
      }
  }
  