content: /**
   * DBSyncer Copyright 2020-2023 All Rights Reserved.
   */
  package org.dbsyncer.sdk.listener;
  
  import org.dbsyncer.common.util.CollectionUtils;
  import org.dbsyncer.common.scheduled.ScheduledTaskService;
  import org.dbsyncer.sdk.config.ListenerConfig;
  import org.dbsyncer.sdk.connector.ConnectorInstance;
  import org.dbsyncer.sdk.constant.ConnectorConstant;
  import org.dbsyncer.sdk.model.ChangedOffset;
  import org.dbsyncer.sdk.model.ConnectorConfig;
  import org.dbsyncer.sdk.model.Table;
  import org.dbsyncer.sdk.spi.ConnectorService;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.sql.Timestamp;
  import java.time.LocalDateTime;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.concurrent.TimeUnit;
  
  /**
   * @version 1.0.0
   * @Author AE86
   * @Date 2020-05-25 22:35
   */
  public abstract class AbstractListener<C extends ConnectorInstance> implements Listener {
  
      private final Logger logger = LoggerFactory.getLogger(getClass());
      private final int FLUSH_DELAYED_SECONDS = 20;
      protected ConnectorInstance connectorInstance;
      protected ConnectorService connectorService;
      protected ScheduledTaskService scheduledTaskService;
      protected ConnectorConfig connectorConfig;
      protected ListenerConfig listenerConfig;
      protected Set<String> filterTable;
      protected List<Table> sourceTable;
      protected Map<String, String> snapshot;
      protected String metaId;
      private Watcher watcher;
  
      @Override
      public void register(Watcher watcher) {
          this.watcher = watcher;
      }
  
      @Override
      public void changeEvent(ChangedEvent event) {
          if (null != event) {
              switch (event.getEvent()) {
                  case ConnectorConstant.OPERTION_UPDATE:
                      // 是否支持监听修改事件
                      processEvent(listenerConfig.isEnableUpdate(), event);
                      break;
                  case ConnectorConstant.OPERTION_INSERT:
                      // 是否支持监听新增事件
                      processEvent(listenerConfig.isEnableInsert(), event);
                      break;
                  case ConnectorConstant.OPERTION_DELETE:
                      // 是否支持监听删除事件
                      processEvent(listenerConfig.isEnableDelete(), event);
                      break;
                  case ConnectorConstant.OPERTION_ALTER:
                      // 表结构变更事件
                      watcher.changeEvent(event);
                      break;
                  default:
                      break;
              }
          }
      }
  
      @Override
      public void refreshEvent(ChangedOffset offset) {
          // nothing to do
      }
  
      @Override
      public void flushEvent() {
          // 20s内更新，执行写入
          if (watcher.getMetaUpdateTime() > Timestamp.valueOf(LocalDateTime.now().minusSeconds(FLUSH_DELAYED_SECONDS)).getTime()) {
              if (!CollectionUtils.isEmpty(snapshot)) {
                  logger.info("snapshot：{}", snapshot);
                  watcher.flushEvent(snapshot);
              }
          }
      }
  
      @Override
      public void forceFlushEvent() {
          if (!CollectionUtils.isEmpty(snapshot)) {
              logger.info("snapshot：{}", snapshot);
              watcher.flushEvent(snapshot);
          }
      }
  
      @Override
      public void errorEvent(Exception e) {
          watcher.errorEvent(e);
      }
  
      protected void sleepInMills(long timeout) {
          try {
              TimeUnit.MILLISECONDS.sleep(timeout);
          } catch (InterruptedException e) {
              logger.info(e.getMessage());
          }
      }
  
      /**
       * 如果允许监听该事件，则触发事件通知
       *
       * @param permitEvent
       * @param event
       */
      private void processEvent(boolean permitEvent, ChangedEvent event) {
          if (permitEvent) {
              watcher.changeEvent(event);
          }
      }
  
      public void setConnectorInstance(ConnectorInstance connectorInstance) {
          this.connectorInstance = connectorInstance;
      }
  
      public void setConnectorService(ConnectorService connectorService) {
          this.connectorService = connectorService;
      }
  
      public C getConnectorInstance() {
          return (C) connectorInstance;
      }
  
      public void setScheduledTaskService(ScheduledTaskService scheduledTaskService) {
          this.scheduledTaskService = scheduledTaskService;
      }
  
      public void setConnectorConfig(ConnectorConfig connectorConfig) {
          this.connectorConfig = connectorConfig;
      }
  
      public void setListenerConfig(ListenerConfig listenerConfig) {
          this.listenerConfig = listenerConfig;
      }
  
      public void setFilterTable(Set<String> filterTable) {
          this.filterTable = filterTable;
      }
  
      public AbstractListener setSourceTable(List<Table> sourceTable) {
          this.sourceTable = sourceTable;
          return this;
      }
  
      public void setSnapshot(Map<String, String> snapshot) {
          this.snapshot = snapshot;
      }
  
      public void setMetaId(String metaId) {
          this.metaId = metaId;
      }
  
  }