content: package com.alibaba.otter.canal.deployer;
  
  import com.alibaba.otter.canal.connector.core.spi.ProxyCanalMQProducer;
  import java.util.Properties;
  
  import com.alibaba.otter.canal.connector.core.config.MQProperties;
  import org.apache.commons.lang.StringUtils;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import com.alibaba.otter.canal.admin.netty.CanalAdminWithNetty;
  import com.alibaba.otter.canal.connector.core.spi.CanalMQProducer;
  import com.alibaba.otter.canal.connector.core.spi.ExtensionLoader;
  import com.alibaba.otter.canal.deployer.admin.CanalAdminController;
  import com.alibaba.otter.canal.server.CanalMQStarter;
  
  /**
   * Canal server 启动类
   *
   * @author rewerma 2020-01-27
   * @version 1.0.2
   */
  public class CanalStarter {
  
      private static final Logger logger = LoggerFactory.getLogger(CanalStarter.class);
  
      private static final String CONNECTOR_SPI_DIR = "/plugin";
      private static final String CONNECTOR_STANDBY_SPI_DIR = "/canal/plugin";
  
      private CanalController controller = null;
      private CanalMQProducer canalMQProducer = null;
      private Thread shutdownThread = null;
      private CanalMQStarter canalMQStarter = null;
      private volatile Properties properties;
      private volatile boolean running = false;
  
      private CanalAdminWithNetty canalAdmin;
  
      public CanalStarter(Properties properties) {
          this.properties = properties;
      }
  
      public boolean isRunning() {
          return running;
      }
  
      public Properties getProperties() {
          return properties;
      }
  
      public void setProperties(Properties properties) {
          this.properties = properties;
      }
  
      public CanalController getController() {
          return controller;
      }
  
      /**
       * 启动方法
       *
       * @throws Throwable
       */
      public synchronized void start() throws Throwable {
          String serverMode = CanalController.getProperty(properties, CanalConstants.CANAL_SERVER_MODE);
          if (!"tcp".equalsIgnoreCase(serverMode)) {
              ExtensionLoader<CanalMQProducer> loader = ExtensionLoader.getExtensionLoader(CanalMQProducer.class);
              canalMQProducer = loader
                  .getExtension(serverMode.toLowerCase(), CONNECTOR_SPI_DIR, CONNECTOR_STANDBY_SPI_DIR);
              if (canalMQProducer != null) {
                  canalMQProducer =  new ProxyCanalMQProducer(canalMQProducer);
                  canalMQProducer.init(properties);
              }
          }
  
          if (canalMQProducer != null) {
              MQProperties mqProperties = canalMQProducer.getMqProperties();
              // disable netty
              System.setProperty(CanalConstants.CANAL_WITHOUT_NETTY, "true");
              if (mqProperties.isFlatMessage()) {
                  // 设置为raw避免ByteString->Entry的二次解析
                  System.setProperty("canal.instance.memory.rawEntry", "false");
              }
          }
  
          logger.info("## start the canal server.");
          controller = new CanalController(properties);
          controller.start();
          logger.info("## the canal server is running now ......");
          shutdownThread = new Thread(() -> {
              try {
                  logger.info("## stop the canal server");
                  controller.stop();
                  CanalLauncher.runningLatch.countDown();
              } catch (Throwable e) {
                  logger.warn("##something goes wrong when stopping canal Server:", e);
              } finally {
                  logger.info("## canal server is down.");
              }
          });
          Runtime.getRuntime().addShutdownHook(shutdownThread);
  
          if (canalMQProducer != null) {
              canalMQStarter = new CanalMQStarter(canalMQProducer);
              String destinations = CanalController.getDestinations(properties);
              canalMQStarter.start(destinations);
              controller.setCanalMQStarter(canalMQStarter);
          }
  
          // start canalAdmin
          String port = CanalController.getProperty(properties, CanalConstants.CANAL_ADMIN_PORT);
          if (canalAdmin == null && StringUtils.isNotEmpty(port)) {
              String user = CanalController.getProperty(properties, CanalConstants.CANAL_ADMIN_USER);
              String passwd = CanalController.getProperty(properties, CanalConstants.CANAL_ADMIN_PASSWD);
              CanalAdminController canalAdmin = new CanalAdminController(this);
              canalAdmin.setUser(user);
              canalAdmin.setPasswd(passwd);
              String ip = CanalController.getProperty(properties, CanalConstants.CANAL_IP);
  
              logger.debug("canal admin port:{}, canal admin user:{}, canal admin password: {}, canal ip:{}",
                  port,
                  user,
                  passwd,
                  ip);
  
              CanalAdminWithNetty canalAdminWithNetty = CanalAdminWithNetty.instance();
              canalAdminWithNetty.setCanalAdmin(canalAdmin);
              canalAdminWithNetty.setPort(Integer.parseInt(port));
              canalAdminWithNetty.setIp(ip);
              canalAdminWithNetty.start();
              this.canalAdmin = canalAdminWithNetty;
          }
  
          running = true;
      }
  
      public synchronized void stop() throws Throwable {
          stop(false);
      }
  
      /**
       * 销毁方法，远程配置变更时调用
       *
       * @throws Throwable
       */
      public synchronized void stop(boolean stopByAdmin) throws Throwable {
          if (!stopByAdmin && canalAdmin != null) {
              canalAdmin.stop();
              canalAdmin = null;
          }
  
          if (controller != null) {
              controller.stop();
              controller = null;
          }
          if (shutdownThread != null) {
              Runtime.getRuntime().removeShutdownHook(shutdownThread);
              shutdownThread = null;
          }
          if (canalMQProducer != null && canalMQStarter != null) {
              canalMQStarter.destroy();
              canalMQStarter = null;
              canalMQProducer = null;
          }
          running = false;
      }
  }
  