add module dbsyncer-connector-elasticsearch

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/ConnectorConfigChecker.java

@@ -1,33 +0,0 @@
-package org.dbsyncer.biz.checker;
-
-import org.dbsyncer.parser.model.Mapping;
-import org.dbsyncer.parser.model.TableGroup;
-
-import java.util.Map;
-
-/**
- * @param <C> ConnectorConfig
- * @author AE86
- * @version 1.0.0
- * @date 2020/1/7 23:17
- */
-public interface ConnectorConfigChecker<C> {
-
-    /**
-     * 处理增量同步策略
-     *
-     * @param mapping
-     * @param tableGroup
-     */
-    default void dealIncrementStrategy(Mapping mapping, TableGroup tableGroup) {}
-
-    /**
-     * 修改配置
-     *
-     * @param connectorConfig
-     * @param params
-     * @return
-     */
-    void modify(C connectorConfig, Map<String, String> params);
-
-}

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/AbstractDataBaseConfigChecker.java → dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/AbstractDataBaseConfigValidator.java

@@ -1,9 +1,12 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.biz.checker.impl.connector;
 
-import org.dbsyncer.biz.checker.ConnectorConfigChecker;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.JsonUtil;
 import org.dbsyncer.sdk.config.DatabaseConfig;
+import org.dbsyncer.sdk.connector.ConfigValidator;
 import org.dbsyncer.sdk.model.SqlTable;
 import org.springframework.util.Assert;
 
@@ -11,11 +14,13 @@ import java.util.List;
 import java.util.Map;
 
 /**
- * @author AE86
- * @version 1.0.0
- * @date 2020/1/8 15:17
+ * 关系型数据库连接配置校验器
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2020-01-08 15:17
  */
-public abstract class AbstractDataBaseConfigChecker implements ConnectorConfigChecker<DatabaseConfig> {
+public abstract class AbstractDataBaseConfigValidator implements ConfigValidator<DatabaseConfig> {
 
     @Override
     public void modify(DatabaseConfig connectorConfig, Map<String, String> params) {

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/ConnectorChecker.java

@@ -1,8 +1,11 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.biz.checker.impl.connector;
 
 import org.dbsyncer.biz.BizException;
 import org.dbsyncer.biz.checker.AbstractChecker;
-import org.dbsyncer.biz.checker.ConnectorConfigChecker;
+import org.dbsyncer.sdk.connector.ConfigValidator;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.ConnectorFactory;
 import org.dbsyncer.parser.ProfileComponent;
@@ -21,9 +24,9 @@ import javax.annotation.Resource;
 import java.util.Map;
 
 /**
- * @author AE86
- * @version 1.0.0
- * @date 2020/1/8 15:17
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2020-01-08 15:17
  */
 @Component
 public class ConnectorChecker extends AbstractChecker {
@@ -37,7 +40,7 @@ public class ConnectorChecker extends AbstractChecker {
     private ConnectorFactory connectorFactory;
 
     @Resource
-    private Map<String, ConnectorConfigChecker> map;
+    private Map<String, ConfigValidator> map;
 
     @Override
     public ConfigModel checkAddConfigModel(Map<String, String> params) {
@@ -54,7 +57,7 @@ public class ConnectorChecker extends AbstractChecker {
 
         // 配置连接器配置
         String type = StringUtil.toLowerCaseFirstOne(connectorType).concat("ConfigChecker");
-        ConnectorConfigChecker checker = map.get(type);
+        ConfigValidator checker = map.get(type);
         Assert.notNull(checker, "Checker can not be null.");
         checker.modify(config, params);
 
@@ -83,7 +86,7 @@ public class ConnectorChecker extends AbstractChecker {
 
         // 配置连接器配置
         String type = StringUtil.toLowerCaseFirstOne(config.getConnectorType()).concat("ConfigChecker");
-        ConnectorConfigChecker checker = map.get(type);
+        ConfigValidator checker = map.get(type);
         Assert.notNull(checker, "Checker can not be null.");
         checker.modify(config, params);
 

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/DqlMySQLConfigChecker.java

@@ -1,3 +1,6 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.biz.checker.impl.connector;
 
 import org.dbsyncer.sdk.config.DatabaseConfig;
@@ -6,12 +9,12 @@ import org.springframework.stereotype.Component;
 import java.util.Map;
 
 /**
- * @author AE86
- * @version 1.0.0
- * @date 2020/1/8 15:17
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2020-01-07 15:17
  */
 @Component
-public class DqlMySQLConfigChecker extends AbstractDataBaseConfigChecker {
+public class DqlMySQLConfigChecker extends AbstractDataBaseConfigValidator {
 
     @Override
     public void modify(DatabaseConfig connectorConfig, Map<String, String> params) {

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/DqlOracleConfigChecker.java

@@ -1,3 +1,6 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.biz.checker.impl.connector;
 
 import org.dbsyncer.common.util.StringUtil;
@@ -7,12 +10,12 @@ import org.springframework.stereotype.Component;
 import java.util.Map;
 
 /**
- * @author AE86
- * @version 1.0.0
- * @date 2020/1/8 15:17
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2020-01-08 15:17
  */
 @Component
-public class DqlOracleConfigChecker extends AbstractDataBaseConfigChecker {
+public class DqlOracleConfigChecker extends AbstractDataBaseConfigValidator {
 
     @Override
     public void modify(DatabaseConfig connectorConfig, Map<String, String> params) {

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/DqlPostgreSQLConfigChecker.java

@@ -1,3 +1,6 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.biz.checker.impl.connector;
 
 import org.dbsyncer.common.util.StringUtil;
@@ -7,12 +10,12 @@ import org.springframework.stereotype.Component;
 import java.util.Map;
 
 /**
- * @author AE86
- * @version 1.0.0
- * @date 2022/4/5 22:14
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2022-04-05 22:14
  */
 @Component
-public class DqlPostgreSQLConfigChecker extends AbstractDataBaseConfigChecker {
+public class DqlPostgreSQLConfigChecker extends AbstractDataBaseConfigValidator {
 
     @Override
     public void modify(DatabaseConfig connectorConfig, Map<String, String> params) {

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/DqlSqlServerConfigChecker.java → dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/DqlSqlServerConfigValidator.java

@@ -1,22 +1,27 @@
-package org.dbsyncer.biz.checker.impl.connector;
-
-import org.dbsyncer.sdk.config.DatabaseConfig;
-import org.springframework.stereotype.Component;
-
-import java.util.Map;
-
-/**
- * @author AE86
- * @version 1.0.0
- * @date 2021/5/18 20:17
- */
-@Component
-public class DqlSqlServerConfigChecker extends AbstractDataBaseConfigChecker {
-
-    @Override
-    public void modify(DatabaseConfig connectorConfig, Map<String, String> params) {
-        super.modify(connectorConfig, params);
-        super.modifyDql(connectorConfig, params);
-        super.modifySchema(connectorConfig, params);
-    }
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.biz.checker.impl.connector;
+
+import org.dbsyncer.sdk.config.DatabaseConfig;
+import org.springframework.stereotype.Component;
+
+import java.util.Map;
+
+/**
+ * DqlSqlServer连接配置校验器实现
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2022-05-18 20:17
+ */
+@Component
+public class DqlSqlServerConfigValidator extends AbstractDataBaseConfigValidator {
+
+    @Override
+    public void modify(DatabaseConfig connectorConfig, Map<String, String> params) {
+        super.modify(connectorConfig, params);
+        super.modifyDql(connectorConfig, params);
+        super.modifySchema(connectorConfig, params);
+    }
 }

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/FileConfigChecker.java → dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/FileConfigValidator.java

@@ -1,6 +1,9 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.biz.checker.impl.connector;
 
-import org.dbsyncer.biz.checker.ConnectorConfigChecker;
+import org.dbsyncer.sdk.connector.ConfigValidator;
 import org.dbsyncer.common.util.JsonUtil;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.config.FileConfig;
@@ -13,12 +16,14 @@ import java.util.List;
 import java.util.Map;
 
 /**
- * @author AE86
- * @version 1.0.0
- * @date 2022/5/6 0:04
+ * 文件连接配置校验器实现
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2022-05-06 00:04
  */
 @Component
-public class FileConfigChecker implements ConnectorConfigChecker<FileConfig> {
+public class FileConfigValidator implements ConfigValidator<FileConfig> {
 
     @Override
     public void modify(FileConfig fileConfig, Map<String, String> params) {

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/KafkaConfigChecker.java → dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/KafkaConfigValidator.java

@@ -1,6 +1,9 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.biz.checker.impl.connector;
 
-import org.dbsyncer.biz.checker.ConnectorConfigChecker;
+import org.dbsyncer.sdk.connector.ConfigValidator;
 import org.dbsyncer.common.util.NumberUtil;
 import org.dbsyncer.connector.config.KafkaConfig;
 import org.springframework.stereotype.Component;
@@ -9,12 +12,14 @@ import org.springframework.util.Assert;
 import java.util.Map;
 
 /**
- * @author AE86
- * @version 1.0.0
- * @date 2021/11/22 23:55
+ * Kafka连接配置校验器实现
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2021-11-22 23:55
  */
 @Component
-public class KafkaConfigChecker implements ConnectorConfigChecker<KafkaConfig> {
+public class KafkaConfigValidator implements ConfigValidator<KafkaConfig> {
 
     @Override
     public void modify(KafkaConfig connectorConfig, Map<String, String> params) {

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/MySQLConfigChecker.java

@@ -1,13 +1,16 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.biz.checker.impl.connector;
 
 import org.springframework.stereotype.Component;
 
 /**
- * @author AE86
- * @version 1.0.0
- * @date 2020/1/8 15:17
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2020-01-08 15:17
  */
 @Component
-public class MySQLConfigChecker extends AbstractDataBaseConfigChecker {
+public class MySQLConfigChecker extends AbstractDataBaseConfigValidator {
 
 }

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/OracleConfigChecker.java

@@ -1,176 +0,0 @@
-package org.dbsyncer.biz.checker.impl.connector;
-
-import org.dbsyncer.biz.enums.OracleIncrementEnum;
-import org.dbsyncer.common.util.CollectionUtils;
-import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.sdk.config.DatabaseConfig;
-import org.dbsyncer.parser.ParserComponent;
-import org.dbsyncer.parser.ProfileComponent;
-import org.dbsyncer.parser.model.FieldMapping;
-import org.dbsyncer.parser.model.Mapping;
-import org.dbsyncer.parser.model.TableGroup;
-import org.dbsyncer.sdk.model.Field;
-import org.dbsyncer.sdk.model.MetaInfo;
-import org.dbsyncer.sdk.model.Table;
-import org.springframework.stereotype.Component;
-import org.springframework.util.Assert;
-
-import javax.annotation.Resource;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
-/**
- * <p>1、增量同步时，目标源必须有一个主键字段用于接收ROW_ID值。</p>
- * <p>2、全局可配置目标源ROW_ID字段名称，默认为ROW_ID_LABEL_NAME。 </p>
- * <p>3、如果配置了接收字段，添加字段映射关系[ROW_ID_LABEL_NAME] > [ROW_ID_LABEL_NAME]，并将ROW_ID_LABEL_NAME字段设置为目标源的唯一主键。</p>
- * <p>4、全量同步时，ROW_ID_LABEL_NAME参数非必须。</p>
- *
- * @author AE86
- * @version 1.0.0
- * @date 2020/1/8 15:17
- */
-@Component
-public class OracleConfigChecker extends AbstractDataBaseConfigChecker {
-
-    @Resource
-    private ParserComponent parserComponent;
-
-    @Resource
-    private ProfileComponent profileComponent;
-
-    @Override
-    public void modify(DatabaseConfig connectorConfig, Map<String, String> params) {
-        super.modify(connectorConfig, params);
-
-        String schema = params.get("schema");
-        connectorConfig.setSchema(StringUtil.isBlank(schema) ? connectorConfig.getUsername().toUpperCase() : schema.toUpperCase());
-    }
-
-    @Override
-    public void dealIncrementStrategy(Mapping mapping, TableGroup tableGroup) {
-        String rowIdLabelName = OracleIncrementEnum.ROW_ID_LABEL_NAME.getName();
-        Map<String, String> params = tableGroup.getParams();
-        if (CollectionUtils.isEmpty(params) || !params.containsKey(rowIdLabelName)) {
-            revert(mapping, tableGroup);
-            return;
-        }
-        String targetRowIdName = params.get(rowIdLabelName);
-
-        // 检查目标源是否支持该自定义字段
-        List<Field> targetColumn = tableGroup.getTargetTable().getColumn();
-        Field targetField = null;
-        for (Field f : targetColumn) {
-            if (StringUtil.equals(f.getName(), targetRowIdName)) {
-                targetField = f;
-                targetField.setPk(true);
-                continue;
-            }
-            f.setPk(false);
-        }
-        Assert.isTrue(null != targetField,
-                String.format("[%s 同步到 %s]，目标源表不存在字段%s", tableGroup.getSourceTable().getName(), tableGroup.getTargetTable().getName(),
-                        targetRowIdName));
-
-        // 检查是否更新
-        List<FieldMapping> fieldMapping = tableGroup.getFieldMapping();
-        List<FieldMapping> list = new ArrayList<>();
-        for (FieldMapping m : fieldMapping) {
-            if (null != m.getSource() && OracleIncrementEnum.isRowId(m.getSource().getName())) {
-                m.getTarget().setName(targetRowIdName);
-                return;
-            }
-
-            // 排除主键关系
-            if (null != m.getTarget() && StringUtil.equals(m.getTarget().getName(), targetRowIdName)) {
-                continue;
-            }
-
-            list.add(m);
-        }
-
-        Field sourceField = new Field(OracleIncrementEnum.ROW_ID.getName(), "VARCHAR2", 12, false,
-                OracleIncrementEnum.ROW_ID_LABEL_NAME.getName(), true);
-        tableGroup.getSourceTable().getColumn().add(0, sourceField);
-
-        list.add(0, new FieldMapping(sourceField, targetField));
-        fieldMapping.clear();
-        fieldMapping.addAll(list);
-    }
-
-    /**
-     * 还原字段和映射关系
-     *
-     * @param mapping
-     * @param tableGroup
-     */
-    private void revert(Mapping mapping, TableGroup tableGroup) {
-        List<FieldMapping> fieldMapping = tableGroup.getFieldMapping();
-        if (CollectionUtils.isEmpty(fieldMapping)) {
-            return;
-        }
-
-        // 还原字段
-        Table sourceTable = tableGroup.getSourceTable();
-        List<Field> sourceColumn = sourceTable.getColumn();
-        List<Field> sourceFields = new ArrayList<>();
-        boolean existRowId = false;
-        for (Field f : sourceColumn) {
-            if (OracleIncrementEnum.isRowId(f.getName())) {
-                existRowId = true;
-                continue;
-            }
-            sourceFields.add(f);
-        }
-        sourceTable.setColumn(sourceFields);
-
-        if (!existRowId) {
-            return;
-        }
-
-        // 存在自定义主键
-        String pk = null;
-        for (FieldMapping m : tableGroup.getFieldMapping()) {
-            if (null != m.getSource() && OracleIncrementEnum.isRowId(m.getSource().getName())) {
-                continue;
-            }
-            if (null != m.getTarget() && m.getTarget().isPk()) {
-                pk = m.getTarget().getName();
-                break;
-            }
-        }
-
-        // 没有自定义主键，获取表元信息
-        if (null == pk) {
-            Table targetTable = tableGroup.getTargetTable();
-            MetaInfo metaInfo = parserComponent.getMetaInfo(mapping.getTargetConnectorId(), targetTable.getName());
-            List<Field> targetColumn = metaInfo.getColumn();
-            targetTable.setColumn(targetColumn);
-
-            if (!CollectionUtils.isEmpty(targetColumn)) {
-                for (Field f : targetColumn) {
-                    if (f.isPk()) {
-                        pk = f.getName();
-                        break;
-                    }
-                }
-            }
-        }
-
-        // 剔除映射关系
-        List<FieldMapping> list = new ArrayList<>();
-        for (FieldMapping m : tableGroup.getFieldMapping()) {
-            if (null != m.getSource() && OracleIncrementEnum.isRowId(m.getSource().getName())) {
-                continue;
-            }
-
-            if (null != m.getTarget() && StringUtil.equals(m.getTarget().getName(), pk)) {
-                m.getTarget().setPk(true);
-            }
-            list.add(m);
-        }
-        fieldMapping.clear();
-        fieldMapping.addAll(list);
-    }
-
-}

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/OracleConfigValidator.java

@@ -0,0 +1,28 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.biz.checker.impl.connector;
+
+import org.dbsyncer.common.util.StringUtil;
+import org.dbsyncer.sdk.config.DatabaseConfig;
+
+import java.util.Map;
+
+/**
+ * Oracle连接配置校验器实现
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2020-01-08 15:17
+ */
+public class OracleConfigValidator extends AbstractDataBaseConfigValidator {
+
+    @Override
+    public void modify(DatabaseConfig connectorConfig, Map<String, String> params) {
+        super.modify(connectorConfig, params);
+
+        String schema = params.get("schema");
+        connectorConfig.setSchema(StringUtil.isBlank(schema) ? connectorConfig.getUsername().toUpperCase() : schema.toUpperCase());
+    }
+
+}

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/PostgreSQLConfigChecker.java

@@ -1,3 +1,6 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.biz.checker.impl.connector;
 
 import org.dbsyncer.common.util.StringUtil;
@@ -7,12 +10,12 @@ import org.springframework.stereotype.Component;
 import java.util.Map;
 
 /**
- * @author AE86
- * @version 1.0.0
- * @date 2022/4/5 22:14
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2022-04-05 22:14
  */
 @Component
-public class PostgreSQLConfigChecker extends AbstractDataBaseConfigChecker {
+public class PostgreSQLConfigChecker extends AbstractDataBaseConfigValidator {
     @Override
     public void modify(DatabaseConfig connectorConfig, Map<String, String> params) {
         super.modify(connectorConfig, params);

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/SqlServerConfigChecker.java

@@ -1,3 +1,6 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.biz.checker.impl.connector;
 
 import org.dbsyncer.sdk.config.DatabaseConfig;
@@ -6,12 +9,12 @@ import org.springframework.stereotype.Component;
 import java.util.Map;
 
 /**
- * @author AE86
- * @version 1.0.0
- * @date 2021/5/18 20:17
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2023-11-25 23:10
  */
 @Component
-public class SqlServerConfigChecker extends AbstractDataBaseConfigChecker {
+public class SqlServerConfigChecker extends AbstractDataBaseConfigValidator {
     @Override
     public void modify(DatabaseConfig connectorConfig, Map<String, String> params) {
         super.modify(connectorConfig, params);

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/tablegroup/TableGroupChecker.java

@@ -1,23 +1,26 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.biz.checker.impl.tablegroup;
 
 import org.dbsyncer.biz.BizException;
 import org.dbsyncer.biz.checker.AbstractChecker;
-import org.dbsyncer.biz.checker.ConnectorConfigChecker;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.JsonUtil;
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.sdk.model.Field;
-import org.dbsyncer.sdk.model.MetaInfo;
-import org.dbsyncer.sdk.model.Table;
-import org.dbsyncer.sdk.util.PrimaryKeyUtil;
 import org.dbsyncer.parser.ParserComponent;
 import org.dbsyncer.parser.ProfileComponent;
-import org.dbsyncer.sdk.enums.ModelEnum;
 import org.dbsyncer.parser.model.ConfigModel;
 import org.dbsyncer.parser.model.FieldMapping;
 import org.dbsyncer.parser.model.Mapping;
 import org.dbsyncer.parser.model.TableGroup;
 import org.dbsyncer.parser.util.PickerUtil;
+import org.dbsyncer.sdk.connector.ConfigValidator;
+import org.dbsyncer.sdk.enums.ModelEnum;
+import org.dbsyncer.sdk.model.Field;
+import org.dbsyncer.sdk.model.MetaInfo;
+import org.dbsyncer.sdk.model.Table;
+import org.dbsyncer.sdk.util.PrimaryKeyUtil;
 import org.dbsyncer.storage.constant.ConfigConstant;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -36,9 +39,9 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
 /**
- * @author AE86
- * @version 1.0.0
- * @date 2020/1/8 15:17
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2020-01-08 15:17
  */
 @Component
 public class TableGroupChecker extends AbstractChecker {
@@ -52,7 +55,7 @@ public class TableGroupChecker extends AbstractChecker {
     private ProfileComponent profileComponent;
 
     @Resource
-    private Map<String, ConnectorConfigChecker> map;
+    private Map<String, ConfigValidator> map;
 
     @Override
     public ConfigModel checkAddConfigModel(Map<String, String> params) {
@@ -139,9 +142,6 @@ public class TableGroupChecker extends AbstractChecker {
         // 合并高级配置
         TableGroup group = PickerUtil.mergeTableGroupConfig(mapping, tableGroup);
 
-        // 处理策略
-        dealIncrementStrategy(mapping, group);
-
         Map<String, String> command = parserComponent.getCommand(mapping, group);
         tableGroup.setCommand(command);
 
@@ -150,14 +150,6 @@ public class TableGroupChecker extends AbstractChecker {
         tableGroup.getSourceTable().setCount(count);
     }
 
-    public void dealIncrementStrategy(Mapping mapping, TableGroup tableGroup) {
-        String connectorType = profileComponent.getConnector(mapping.getSourceConnectorId()).getConfig().getConnectorType();
-        String type = StringUtil.toLowerCaseFirstOne(connectorType).concat("ConfigChecker");
-        ConnectorConfigChecker checker = map.get(type);
-        Assert.notNull(checker, "Checker can not be null.");
-        checker.dealIncrementStrategy(mapping, tableGroup);
-    }
-
     private Table getTable(String connectorId, String tableName, String primaryKeyStr) {
         MetaInfo metaInfo = parserComponent.getMetaInfo(connectorId, tableName);
         Assert.notNull(metaInfo, "无法获取连接器表信息:" + tableName);

dbsyncer-biz/src/main/java/org/dbsyncer/biz/impl/ConditionServiceImpl.java

@@ -1,8 +1,11 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.biz.impl;
 
 import org.dbsyncer.biz.ConditionService;
 import org.dbsyncer.biz.vo.ConditionVo;
-import org.dbsyncer.connector.enums.FilterEnum;
+import org.dbsyncer.sdk.enums.FilterEnum;
 import org.dbsyncer.sdk.enums.QuartzFilterEnum;
 import org.dbsyncer.parser.ProfileComponent;
 import org.dbsyncer.sdk.enums.OperationEnum;
@@ -12,9 +15,11 @@ import javax.annotation.Resource;
 import java.util.List;
 
 /**
- * @author AE86
- * @version 1.0.0
- * @date 2020/1/19 16:02
+ * 支持的条件和运算符类型
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2020-01-19 16:02
  */
 @Component
 public class ConditionServiceImpl implements ConditionService {

dbsyncer-biz/src/main/java/org/dbsyncer/biz/impl/MonitorServiceImpl.java

@@ -1,3 +1,6 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.biz.impl;
 
 import org.dbsyncer.biz.DataSyncService;
@@ -25,7 +28,7 @@ import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.JsonUtil;
 import org.dbsyncer.common.util.NumberUtil;
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.enums.FilterEnum;
+import org.dbsyncer.sdk.enums.FilterEnum;
 import org.dbsyncer.sdk.scheduled.ScheduledTaskJob;
 import org.dbsyncer.sdk.scheduled.ScheduledTaskService;
 import org.dbsyncer.parser.LogService;
@@ -62,9 +65,9 @@ import java.util.Map;
 import java.util.stream.Collectors;
 
 /**
- * @author AE86
- * @version 1.0.0
- * @date 2020/04/27 10:20
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2020-04-27 10:20
  */
 @Service
 public class MonitorServiceImpl extends BaseServiceImpl implements MonitorService, ScheduledTaskJob {

dbsyncer-biz/src/main/java/org/dbsyncer/biz/vo/ConditionVo.java

@@ -1,15 +1,18 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.biz.vo;
 
-import org.dbsyncer.connector.enums.FilterEnum;
+import org.dbsyncer.sdk.enums.FilterEnum;
 import org.dbsyncer.sdk.enums.QuartzFilterEnum;
 import org.dbsyncer.sdk.enums.OperationEnum;
 
 import java.util.List;
 
 /**
- * @author AE86
- * @version 1.0.0
- * @date 2020/01/19 17:20
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2020-01-19 17:20
  */
 public class ConditionVo {
 

dbsyncer-connector/dbsyncer-connector-elasticsearch/pom.xml

@@ -0,0 +1,42 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>dbsyncer-connector</artifactId>
+        <groupId>org.ghi</groupId>
+        <version>2.0.0</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>dbsyncer-connector-elasticsearch</artifactId>
+
+    <dependencies>
+        <!-- sdk -->
+        <dependency>
+            <groupId>org.ghi</groupId>
+            <artifactId>dbsyncer-sdk</artifactId>
+            <version>${project.parent.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <!-- elasticsearch -->
+        <dependency>
+            <groupId>org.elasticsearch.client</groupId>
+            <artifactId>elasticsearch-rest-high-level-client</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-log4j2</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+    </dependencies>
+</project>

dbsyncer-connector/src/main/java/org/dbsyncer/connector/es/ESConnectorInstance.java → dbsyncer-connector/dbsyncer-connector-elasticsearch/src/main/java/org/dbsyncer/connector/elasticsearch/ESConnectorInstance.java

@@ -1,13 +1,24 @@
-package org.dbsyncer.connector.es;
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.connector.elasticsearch;
 
-import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.config.ESConfig;
-import org.dbsyncer.connector.util.ESUtil;
+import org.dbsyncer.connector.elasticsearch.api.EasyRestHighLevelClient;
+import org.dbsyncer.connector.elasticsearch.config.ESConfig;
+import org.dbsyncer.connector.elasticsearch.util.ESUtil;
 import org.dbsyncer.sdk.connector.ConnectorInstance;
+import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.Version;
 import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.core.MainResponse;
 
+/**
+ * ES连接器实例
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2023-11-25 23:10
+ */
 public final class ESConnectorInstance implements ConnectorInstance<ESConfig, EasyRestHighLevelClient> {
     private ESConfig config;
     private EasyRestHighLevelClient client;
@@ -19,7 +30,7 @@ public final class ESConnectorInstance implements ConnectorInstance<ESConfig, Ea
             MainResponse info = client.info(RequestOptions.DEFAULT);
             client.setVersion(Version.fromString(info.getVersion().getNumber()));
         } catch (Exception e) {
-            throw new ConnectorException(String.format("获取ES版本信息异常 %s, %s", config.getUrl(), e.getMessage()));
+            throw new ElasticsearchException(String.format("获取ES版本信息异常 %s, %s", config.getUrl(), e.getMessage()));
         }
     }
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/es/ESConnector.java → dbsyncer-connector/dbsyncer-connector-elasticsearch/src/main/java/org/dbsyncer/connector/elasticsearch/ElasticsearchConnector.java

@@ -1,20 +1,27 @@
-package org.dbsyncer.connector.es;
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.connector.elasticsearch;
 
 import org.dbsyncer.common.model.Result;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.JsonUtil;
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.config.ESConfig;
-import org.dbsyncer.connector.enums.ESFieldTypeEnum;
-import org.dbsyncer.connector.enums.FilterEnum;
-import org.dbsyncer.connector.util.ESUtil;
+import org.dbsyncer.connector.elasticsearch.cdc.ESQuartzListener;
+import org.dbsyncer.connector.elasticsearch.config.ESConfig;
+import org.dbsyncer.connector.elasticsearch.enums.ESFieldTypeEnum;
+import org.dbsyncer.connector.elasticsearch.schema.ESDateValueMapper;
+import org.dbsyncer.connector.elasticsearch.schema.ESOtherValueMapper;
+import org.dbsyncer.connector.elasticsearch.util.ESUtil;
+import org.dbsyncer.connector.elasticsearch.validator.ESConfigValidator;
 import org.dbsyncer.sdk.config.CommandConfig;
 import org.dbsyncer.sdk.config.ReaderConfig;
 import org.dbsyncer.sdk.config.WriterBatchConfig;
 import org.dbsyncer.sdk.connector.AbstractConnector;
+import org.dbsyncer.sdk.connector.ConfigValidator;
 import org.dbsyncer.sdk.connector.ConnectorInstance;
 import org.dbsyncer.sdk.constant.ConnectorConstant;
+import org.dbsyncer.sdk.enums.FilterEnum;
 import org.dbsyncer.sdk.enums.ListenerTypeEnum;
 import org.dbsyncer.sdk.enums.OperationEnum;
 import org.dbsyncer.sdk.listener.Listener;
@@ -24,6 +31,7 @@ import org.dbsyncer.sdk.model.MetaInfo;
 import org.dbsyncer.sdk.model.Table;
 import org.dbsyncer.sdk.spi.ConnectorService;
 import org.dbsyncer.sdk.util.PrimaryKeyUtil;
+import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.bulk.BulkResponse;
@@ -49,7 +57,6 @@ import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.search.sort.SortOrder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.stereotype.Component;
 
 import javax.annotation.PostConstruct;
 import java.io.IOException;
@@ -62,8 +69,14 @@ import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 
-@Component
-public final class ESConnector extends AbstractConnector implements ConnectorService<ESConnectorInstance, ESConfig> {
+/**
+ * ES连接器实现
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2023-11-25 23:10
+ */
+public final class ElasticsearchConnector extends AbstractConnector implements ConnectorService<ESConnectorInstance, ESConfig> {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
@@ -71,6 +84,8 @@ public final class ESConnector extends AbstractConnector implements ConnectorSer
 
     private final Map<String, FilterMapper> filters = new LinkedHashMap<>();
 
+    private final ESConfigValidator configValidator = new ESConfigValidator();
+
     @PostConstruct
     private void init() {
         VALUE_MAPPERS.put(Types.DATE, new ESDateValueMapper());
@@ -110,6 +125,11 @@ public final class ESConnector extends AbstractConnector implements ConnectorSer
         return new ESConnectorInstance(config);
     }
 
+    @Override
+    public ConfigValidator getConfigValidator() {
+        return configValidator;
+    }
+
     @Override
     public void disconnect(ESConnectorInstance connectorInstance) {
         connectorInstance.close();
@@ -122,7 +142,7 @@ public final class ESConnector extends AbstractConnector implements ConnectorSer
             return client.ping(RequestOptions.DEFAULT);
         } catch (IOException e) {
             logger.error(e.getMessage());
-            throw new ConnectorException(e.getMessage());
+            throw new ElasticsearchException(e.getMessage());
         }
     }
 
@@ -151,7 +171,7 @@ public final class ESConnector extends AbstractConnector implements ConnectorSer
             return tables;
         } catch (IOException e) {
             logger.error(e.getMessage());
-            throw new ConnectorException(e);
+            throw new ElasticsearchException(e);
         }
     }
 
@@ -175,7 +195,7 @@ public final class ESConnector extends AbstractConnector implements ConnectorSer
             return new MetaInfo().setColumn(fields);
         } catch (IOException e) {
             logger.error(e.getMessage());
-            throw new ConnectorException(e);
+            throw new ElasticsearchException(e);
         }
     }
 
@@ -196,7 +216,7 @@ public final class ESConnector extends AbstractConnector implements ConnectorSer
             return response.getHits().getTotalHits().value;
         } catch (IOException e) {
             logger.error(e.getMessage());
-            throw new ConnectorException(e);
+            throw new ElasticsearchException(e);
         }
     }
 
@@ -230,7 +250,7 @@ public final class ESConnector extends AbstractConnector implements ConnectorSer
             return new Result(list);
         } catch (IOException e) {
             logger.error(e.getMessage());
-            throw new ConnectorException(e.getMessage());
+            throw new ElasticsearchException(e.getMessage());
         }
     }
 
@@ -239,7 +259,7 @@ public final class ESConnector extends AbstractConnector implements ConnectorSer
         List<Map> data = config.getData();
         if (CollectionUtils.isEmpty(data) || CollectionUtils.isEmpty(config.getFields())) {
             logger.error("writer data can not be empty.");
-            throw new ConnectorException("writer data can not be empty.");
+            throw new ElasticsearchException("writer data can not be empty.");
         }
 
         final Result result = new Result();
@@ -254,7 +274,7 @@ public final class ESConnector extends AbstractConnector implements ConnectorSer
             BulkResponse response = connectorInstance.getConnection().bulkWithVersion(request, RequestOptions.DEFAULT);
             RestStatus restStatus = response.status();
             if (restStatus.getStatus() != RestStatus.OK.getStatus()) {
-                throw new ConnectorException(String.format("error code:%s", restStatus.getStatus()));
+                throw new ElasticsearchException(String.format("error code:%s", restStatus.getStatus()));
             }
             result.addSuccessData(data);
         } catch (Exception e) {
@@ -302,7 +322,7 @@ public final class ESConnector extends AbstractConnector implements ConnectorSer
     private void parseProperties(List<Field> fields, Map<String, Object> sourceMap) {
         Map<String, Object> properties = (Map<String, Object>) sourceMap.get(ESUtil.PROPERTIES);
         if (CollectionUtils.isEmpty(properties)) {
-            throw new ConnectorException("查询字段不能为空.");
+            throw new ElasticsearchException("查询字段不能为空.");
         }
         properties.forEach((fieldName, c) -> {
             Map fieldDesc = (Map) c;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/es/EasyRestHighLevelClient.java → dbsyncer-connector/dbsyncer-connector-elasticsearch/src/main/java/org/dbsyncer/connector/elasticsearch/api/EasyRestHighLevelClient.java

@@ -1,4 +1,7 @@
-package org.dbsyncer.connector.es;
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.connector.elasticsearch.api;
 
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
@@ -22,7 +25,14 @@ import java.util.List;
 
 import static java.util.Collections.emptySet;
 
-public class EasyRestHighLevelClient extends RestHighLevelClient {
+/**
+ * ES客户端（同时兼容6-7版本API）
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2023-11-25 23:10
+ */
+public final class EasyRestHighLevelClient extends RestHighLevelClient {
 
     private Version version;
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/es/RequestConverters.java → dbsyncer-connector/dbsyncer-connector-elasticsearch/src/main/java/org/dbsyncer/connector/elasticsearch/api/RequestConverters.java

@@ -1,4 +1,7 @@
-package org.dbsyncer.connector.es;
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.connector.elasticsearch.api;
 
 import org.apache.http.HttpEntity;
 import org.apache.http.client.methods.HttpDelete;
@@ -81,6 +84,13 @@ import java.util.Locale;
 import java.util.Map;
 import java.util.StringJoiner;
 
+/**
+ * ES请求（同时兼容6-7版本API）
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2023-11-25 23:10
+ */
 final class RequestConverters {
     static final XContentType REQUEST_BODY_CONTENT_TYPE = XContentType.JSON;
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/es/RethrottleRequest.java → dbsyncer-connector/dbsyncer-connector-elasticsearch/src/main/java/org/dbsyncer/connector/elasticsearch/api/RethrottleRequest.java

@@ -1,4 +1,7 @@
-package org.dbsyncer.connector.es;
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.connector.elasticsearch.api;
 
 import org.elasticsearch.client.Validatable;
 import org.elasticsearch.tasks.TaskId;
@@ -7,6 +10,10 @@ import java.util.Objects;
 
 /**
  * A request changing throttling of a task.
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2023-11-25 23:10
  */
 public class RethrottleRequest implements Validatable {
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/es/ESQuartzListener.java → dbsyncer-connector/dbsyncer-connector-elasticsearch/src/main/java/org/dbsyncer/connector/elasticsearch/cdc/ESQuartzListener.java

@@ -1,15 +1,18 @@
-package org.dbsyncer.connector.es;
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.connector.elasticsearch.cdc;
 
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.JsonUtil;
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.ConnectorException;
+import org.dbsyncer.sdk.constant.ConnectorConstant;
 import org.dbsyncer.sdk.enums.QuartzFilterEnum;
 import org.dbsyncer.sdk.listener.AbstractQuartzListener;
-import org.dbsyncer.sdk.model.Point;
 import org.dbsyncer.sdk.listener.QuartzFilter;
-import org.dbsyncer.sdk.constant.ConnectorConstant;
 import org.dbsyncer.sdk.model.Filter;
+import org.dbsyncer.sdk.model.Point;
+import org.elasticsearch.ElasticsearchException;
 
 import java.util.ArrayList;
 import java.util.HashSet;
@@ -18,10 +21,10 @@ import java.util.Map;
 import java.util.Set;
 
 /**
- * ES定时抽取
+ * CDC-ES定时监听器
  *
- * @version 1.0.0
  * @Author AE86
+ * @Version 1.0.0
  * @Date 2021-09-01 20:35
  */
 public final class ESQuartzListener extends AbstractQuartzListener {
@@ -43,7 +46,7 @@ public final class ESQuartzListener extends AbstractQuartzListener {
         Set<String> set = new HashSet<>();
         for (Filter f : filters) {
             if (set.contains(f.getValue())) {
-                throw new ConnectorException(String.format("系统参数%s存在多个.", f.getValue()));
+                throw new ElasticsearchException(String.format("系统参数%s存在多个.", f.getValue()));
             }
             QuartzFilterEnum filterEnum = QuartzFilterEnum.getQuartzFilterEnum(f.getValue());
             if (null != filterEnum) {

dbsyncer-connector/src/main/java/org/dbsyncer/connector/config/ESConfig.java → dbsyncer-connector/dbsyncer-connector-elasticsearch/src/main/java/org/dbsyncer/connector/elasticsearch/config/ESConfig.java

@@ -1,65 +1,69 @@
-package org.dbsyncer.connector.config;
-
-import org.dbsyncer.sdk.model.ConnectorConfig;
-
-/**
- * @author AE86
- * @ClassName: ESConfig
- * @Description: ES连接配置
- * @date: 2021年8月23日 下午8:00:00
- */
-public class ESConfig extends ConnectorConfig {
-
-    /**
-     * 集群地址, http(s)-9200, tcp-9300 http://192.168.1.100:9200,http://192.168.1.200:9200
-     */
-    private String url;
-
-    /**
-     * 帐号
-     */
-    private String username;
-
-    /**
-     * 密码
-     */
-    private String password;
-
-    /**
-     * 索引(相当于数据库)
-     */
-    private String index;
-
-    public String getUrl() {
-        return url;
-    }
-
-    public void setUrl(String url) {
-        this.url = url;
-    }
-
-    public String getUsername() {
-        return username;
-    }
-
-    public void setUsername(String username) {
-        this.username = username;
-    }
-
-    public String getPassword() {
-        return password;
-    }
-
-    public void setPassword(String password) {
-        this.password = password;
-    }
-
-    public String getIndex() {
-        return index;
-    }
-
-    public void setIndex(String index) {
-        this.index = index;
-    }
-
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.connector.elasticsearch.config;
+
+import org.dbsyncer.sdk.model.ConnectorConfig;
+
+/**
+ * ES连接配置
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2021-08-23 20:00
+ */
+public class ESConfig extends ConnectorConfig {
+
+    /**
+     * 集群地址, http(s)-9200, tcp-9300 http://192.168.1.100:9200,http://192.168.1.200:9200
+     */
+    private String url;
+
+    /**
+     * 帐号
+     */
+    private String username;
+
+    /**
+     * 密码
+     */
+    private String password;
+
+    /**
+     * 索引(相当于数据库)
+     */
+    private String index;
+
+    public String getUrl() {
+        return url;
+    }
+
+    public void setUrl(String url) {
+        this.url = url;
+    }
+
+    public String getUsername() {
+        return username;
+    }
+
+    public void setUsername(String username) {
+        this.username = username;
+    }
+
+    public String getPassword() {
+        return password;
+    }
+
+    public void setPassword(String password) {
+        this.password = password;
+    }
+
+    public String getIndex() {
+        return index;
+    }
+
+    public void setIndex(String index) {
+        this.index = index;
+    }
+
 }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/enums/ESFieldTypeEnum.java → dbsyncer-connector/dbsyncer-connector-elasticsearch/src/main/java/org/dbsyncer/connector/elasticsearch/enums/ESFieldTypeEnum.java

@@ -1,7 +1,10 @@
-package org.dbsyncer.connector.enums;
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.connector.elasticsearch.enums;
 
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.ConnectorException;
+import org.elasticsearch.ElasticsearchException;
 
 import java.sql.Types;
 
@@ -10,7 +13,7 @@ import java.sql.Types;
  *
  * @author AE86
  * @version 1.0.0
- * @date 2021/08/26 21:13
+ * @Date 2023-08-26 21:13
  */
 public enum ESFieldTypeEnum {
 
@@ -80,7 +83,7 @@ public enum ESFieldTypeEnum {
         this.type = type;
     }
 
-    public static int getType(String code) throws ConnectorException {
+    public static int getType(String code) throws ElasticsearchException {
         for (ESFieldTypeEnum e : ESFieldTypeEnum.values()) {
             if (StringUtil.equals(e.getCode(), code)) {
                 return e.getType();

dbsyncer-connector/src/main/java/org/dbsyncer/connector/es/ESDateValueMapper.java → dbsyncer-connector/dbsyncer-connector-elasticsearch/src/main/java/org/dbsyncer/connector/elasticsearch/schema/ESDateValueMapper.java

@@ -1,31 +1,36 @@
-package org.dbsyncer.connector.es;
-
-import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.sdk.connector.AbstractValueMapper;
-import org.dbsyncer.sdk.connector.ConnectorInstance;
-
-import java.sql.Date;
-import java.sql.Timestamp;
-
-/**
- * @author moyu
- * @version 1.0.0
- * @date 2023/10/12 0:07
- */
-public class ESDateValueMapper extends AbstractValueMapper<java.util.Date> {
-
-    @Override
-    protected java.util.Date convert(ConnectorInstance connectorInstance, Object val) {
-        if (val instanceof Timestamp) {
-            Timestamp timestamp = (Timestamp) val;
-            return new java.util.Date(timestamp.getTime());
-        }
-
-        if (val instanceof Date) {
-            Date date = (Date) val;
-            return new java.util.Date(date.getTime());
-        }
-
-        throw new ConnectorException(String.format("%s can not find type [%s], val [%s]", getClass().getSimpleName(), val.getClass(), val));
-    }
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.connector.elasticsearch.schema;
+
+import org.dbsyncer.sdk.connector.AbstractValueMapper;
+import org.dbsyncer.sdk.connector.ConnectorInstance;
+import org.elasticsearch.ElasticsearchException;
+
+import java.sql.Date;
+import java.sql.Timestamp;
+
+/**
+ * ES日期字段值转换
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2023-11-25 23:10
+ */
+public class ESDateValueMapper extends AbstractValueMapper<java.util.Date> {
+
+    @Override
+    protected java.util.Date convert(ConnectorInstance connectorInstance, Object val) {
+        if (val instanceof Timestamp) {
+            Timestamp timestamp = (Timestamp) val;
+            return new java.util.Date(timestamp.getTime());
+        }
+
+        if (val instanceof Date) {
+            Date date = (Date) val;
+            return new java.util.Date(date.getTime());
+        }
+
+        throw new ElasticsearchException(String.format("%s can not find type [%s], val [%s]", getClass().getSimpleName(), val.getClass(), val));
+    }
 }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/es/ESOtherValueMapper.java → dbsyncer-connector/dbsyncer-connector-elasticsearch/src/main/java/org/dbsyncer/connector/elasticsearch/schema/ESOtherValueMapper.java

@@ -1,35 +1,40 @@
-package org.dbsyncer.connector.es;
-
-import org.dbsyncer.common.util.JsonUtil;
-import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.sdk.connector.AbstractValueMapper;
-import org.dbsyncer.sdk.connector.ConnectorInstance;
-import org.postgresql.util.PGobject;
-
-import java.util.Map;
-
-/**
- * @author moyu
- * @version 1.0.0
- * @date 2023/10/12 0:07
- */
-public class ESOtherValueMapper extends AbstractValueMapper<Map> {
-
-    @Override
-    protected Map convert(ConnectorInstance connectorInstance, Object val) {
-        if (val instanceof String) {
-            return JsonUtil.jsonToObj((String) val, Map.class);
-        }
-
-        if (val instanceof byte[]) {
-            return JsonUtil.jsonToObj(new String((byte[]) val), Map.class);
-        }
-
-        if (val instanceof PGobject) {
-            PGobject pgObject = (PGobject) val;
-            return JsonUtil.jsonToObj(pgObject.getValue(), Map.class);
-        }
-
-        throw new ConnectorException(String.format("%s can not find type [%s], val [%s]", getClass().getSimpleName(), val.getClass(), val));
-    }
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.connector.elasticsearch.schema;
+
+import org.dbsyncer.common.util.JsonUtil;
+import org.dbsyncer.sdk.connector.AbstractValueMapper;
+import org.dbsyncer.sdk.connector.ConnectorInstance;
+import org.elasticsearch.ElasticsearchException;
+import org.postgresql.util.PGobject;
+
+import java.util.Map;
+
+/**
+ * ES异构字段值转换
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2023-11-25 23:10
+ */
+public class ESOtherValueMapper extends AbstractValueMapper<Map> {
+
+    @Override
+    protected Map convert(ConnectorInstance connectorInstance, Object val) {
+        if (val instanceof String) {
+            return JsonUtil.jsonToObj((String) val, Map.class);
+        }
+
+        if (val instanceof byte[]) {
+            return JsonUtil.jsonToObj(new String((byte[]) val), Map.class);
+        }
+
+        if (val instanceof PGobject) {
+            PGobject pgObject = (PGobject) val;
+            return JsonUtil.jsonToObj(pgObject.getValue(), Map.class);
+        }
+
+        throw new ElasticsearchException(String.format("%s can not find type [%s], val [%s]", getClass().getSimpleName(), val.getClass(), val));
+    }
 }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/util/ESUtil.java → dbsyncer-connector/dbsyncer-connector-elasticsearch/src/main/java/org/dbsyncer/connector/elasticsearch/util/ESUtil.java

@@ -1,62 +1,72 @@
-package org.dbsyncer.connector.util;
-
-import org.apache.http.HttpHost;
-import org.apache.http.auth.AuthScope;
-import org.apache.http.auth.UsernamePasswordCredentials;
-import org.apache.http.client.CredentialsProvider;
-import org.apache.http.conn.ssl.NoopHostnameVerifier;
-import org.apache.http.conn.ssl.TrustAllStrategy;
-import org.apache.http.impl.client.BasicCredentialsProvider;
-import org.apache.http.nio.conn.ssl.SSLIOSessionStrategy;
-import org.apache.http.ssl.SSLContextBuilder;
-import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.config.ESConfig;
-import org.dbsyncer.connector.es.EasyRestHighLevelClient;
-import org.elasticsearch.client.RequestOptions;
-import org.elasticsearch.client.RestClient;
-import org.elasticsearch.client.RestClientBuilder;
-import org.elasticsearch.client.RestHighLevelClient;
-
-import javax.net.ssl.SSLContext;
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.Objects;
-
-public abstract class ESUtil {
-
-    public static final String PROPERTIES = "properties";
-
-    private ESUtil() {
-    }
-
-    public static EasyRestHighLevelClient getConnection(ESConfig config) {
-        String[] ipAddress = StringUtil.split(config.getUrl(), ",");
-        HttpHost[] hosts = Arrays.stream(ipAddress).map(node -> HttpHost.create(node)).filter(Objects::nonNull).toArray(
-                HttpHost[]::new);
-        RestClientBuilder builder = RestClient.builder(hosts);
-        CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
-        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(config.getUsername(), config.getPassword()));
-        try {
-            SSLContext sslContext = new SSLContextBuilder().loadTrustMaterial(new TrustAllStrategy()).build();
-            SSLIOSessionStrategy sessionStrategy = new SSLIOSessionStrategy(sslContext, NoopHostnameVerifier.INSTANCE);
-            builder.setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider).setSSLStrategy(sessionStrategy));
-            final EasyRestHighLevelClient client = new EasyRestHighLevelClient(builder);
-            client.ping(RequestOptions.DEFAULT);
-            return client;
-        } catch (Exception e) {
-            throw new ConnectorException(String.format("Failed to connect to ElasticSearch on %s, %s", config.getUrl(), e.getMessage()));
-        }
-    }
-
-    public static void close(RestHighLevelClient client) {
-        if (null != client) {
-            try {
-                client.close();
-            } catch (IOException e) {
-                throw new ConnectorException(e.getMessage());
-            }
-        }
-    }
-
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.connector.elasticsearch.util;
+
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.conn.ssl.NoopHostnameVerifier;
+import org.apache.http.conn.ssl.TrustAllStrategy;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.nio.conn.ssl.SSLIOSessionStrategy;
+import org.apache.http.ssl.SSLContextBuilder;
+import org.dbsyncer.common.util.StringUtil;
+import org.dbsyncer.connector.elasticsearch.api.EasyRestHighLevelClient;
+import org.dbsyncer.connector.elasticsearch.config.ESConfig;
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.client.RequestOptions;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestClientBuilder;
+import org.elasticsearch.client.RestHighLevelClient;
+
+import javax.net.ssl.SSLContext;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Objects;
+
+/**
+ * ES连接器工具类
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2023-11-25 23:10
+ */
+public abstract class ESUtil {
+
+    public static final String PROPERTIES = "properties";
+
+    private ESUtil() {
+    }
+
+    public static EasyRestHighLevelClient getConnection(ESConfig config) {
+        String[] ipAddress = StringUtil.split(config.getUrl(), ",");
+        HttpHost[] hosts = Arrays.stream(ipAddress).map(node -> HttpHost.create(node)).filter(Objects::nonNull).toArray(
+                HttpHost[]::new);
+        RestClientBuilder builder = RestClient.builder(hosts);
+        CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
+        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(config.getUsername(), config.getPassword()));
+        try {
+            SSLContext sslContext = new SSLContextBuilder().loadTrustMaterial(new TrustAllStrategy()).build();
+            SSLIOSessionStrategy sessionStrategy = new SSLIOSessionStrategy(sslContext, NoopHostnameVerifier.INSTANCE);
+            builder.setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider).setSSLStrategy(sessionStrategy));
+            final EasyRestHighLevelClient client = new EasyRestHighLevelClient(builder);
+            client.ping(RequestOptions.DEFAULT);
+            return client;
+        } catch (Exception e) {
+            throw new ElasticsearchException(String.format("Failed to connect to ElasticSearch on %s, %s", config.getUrl(), e.getMessage()));
+        }
+    }
+
+    public static void close(RestHighLevelClient client) {
+        if (null != client) {
+            try {
+                client.close();
+            } catch (IOException e) {
+                throw new ElasticsearchException(e.getMessage());
+            }
+        }
+    }
+
 }

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/ElasticsearchConfigChecker.java → dbsyncer-connector/dbsyncer-connector-elasticsearch/src/main/java/org/dbsyncer/connector/elasticsearch/validator/ESConfigValidator.java

@@ -1,19 +1,22 @@
-package org.dbsyncer.biz.checker.impl.connector;
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.connector.elasticsearch.validator;
 
-import org.dbsyncer.biz.checker.ConnectorConfigChecker;
-import org.dbsyncer.connector.config.ESConfig;
-import org.springframework.stereotype.Component;
+import org.dbsyncer.connector.elasticsearch.config.ESConfig;
+import org.dbsyncer.sdk.connector.ConfigValidator;
 import org.springframework.util.Assert;
 
 import java.util.Map;
 
 /**
- * @author AE86
- * @version 1.0.0
- * @date 2021/8/25 23:30
+ * ES连接配置校验器实现
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2021-08-25 23:30
  */
-@Component
-public class ElasticsearchConfigChecker implements ConnectorConfigChecker<ESConfig> {
+public final class ESConfigValidator implements ConfigValidator<ESConfig> {
 
     @Override
     public void modify(ESConfig connectorConfig, Map<String, String> params) {

dbsyncer-connector/dbsyncer-connector-elasticsearch/src/main/resources/META-INF/services/org.dbsyncer.sdk.spi.ConnectorService

@@ -0,0 +1 @@
+org.dbsyncer.connector.elasticsearch.ElasticsearchConnector

dbsyncer-connector/src/main/test/ESClientTest.java → dbsyncer-connector/dbsyncer-connector-elasticsearch/src/main/test/ESClientTest.java

@@ -1,5 +1,9 @@
-import org.dbsyncer.connector.config.ESConfig;
-import org.dbsyncer.connector.util.ESUtil;
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+
+import org.dbsyncer.connector.elasticsearch.config.ESConfig;
+import org.dbsyncer.connector.elasticsearch.util.ESUtil;
 import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
 import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.bulk.BulkRequest;
@@ -39,6 +43,13 @@ import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
 
+/**
+ * ES测试用例
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2023-11-25 23:10
+ */
 public class ESClientTest {
     private final Logger logger = LoggerFactory.getLogger(getClass());
     private RestHighLevelClient client;

dbsyncer-connector/pom.xml

@@ -9,6 +9,10 @@
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>dbsyncer-connector</artifactId>
+    <packaging>pom</packaging>
+    <modules>
+        <module>dbsyncer-connector-elasticsearch</module>
+    </modules>
 
     <dependencies>
         <!-- sdk -->
@@ -18,6 +22,13 @@
             <version>${project.parent.version}</version>
         </dependency>
 
+        <!-- dbsyncer-connector-elasticsearch -->
+        <dependency>
+            <groupId>org.ghi</groupId>
+            <artifactId>dbsyncer-connector-elasticsearch</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
+
         <!-- Spring-JDBC -->
         <dependency>
             <groupId>org.springframework</groupId>

dbsyncer-connector/src/main/java/org/dbsyncer/connector/CompareFilter.java

@@ -1,7 +0,0 @@
-package org.dbsyncer.connector;
-
-public interface CompareFilter {
-
-    boolean compare(String value, String filterValue);
-
-}

dbsyncer-connector/src/main/java/org/dbsyncer/connector/ConnectorConfiguration.java

@@ -1,19 +1,47 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.connector;
 
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
+import org.dbsyncer.sdk.spi.ConnectorService;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.BeanDefinitionRegistry;
+import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor;
+import org.springframework.stereotype.Service;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+
+import java.beans.Introspector;
+import java.util.ServiceLoader;
 
 /**
- * @author AE86
- * @version 1.0.0
- * @date 2019/9/19 23:17
+ * 连接器配置
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2019-09-19 23:17
  */
-@Configuration
-public class ConnectorConfiguration {
+@Service
+public class ConnectorConfiguration implements BeanDefinitionRegistryPostProcessor {
 
-    @Bean
-    public ConnectorFactory connectorFactory() {
-        return new ConnectorFactory();
+    @Override
+    public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry beanDefinitionRegistry) throws BeansException {
+        ServiceLoader<ConnectorService> services = ServiceLoader.load(ConnectorService.class, Thread.currentThread().getContextClassLoader());
+        for (ConnectorService s : services) {
+            BeanDefinition beanDefinition = BeanDefinitionBuilder.genericBeanDefinition(s.getClass()).getBeanDefinition();
+            String beanClassName = beanDefinition.getBeanClassName();
+            Assert.state(beanClassName != null, "No bean class name set");
+            String shortClassName = ClassUtils.getShortName(beanClassName);
+            String decapitalize = Introspector.decapitalize(shortClassName);
+            beanDefinitionRegistry.registerBeanDefinition(decapitalize, beanDefinition);
+        }
     }
 
+    @Override
+    public void postProcessBeanFactory(ConfigurableListableBeanFactory configurableListableBeanFactory) throws BeansException {
+
+    }
 }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/ConnectorFactory.java

@@ -1,3 +1,6 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.connector;
 
 import org.dbsyncer.common.model.Result;
@@ -15,6 +18,7 @@ import org.dbsyncer.sdk.model.Table;
 import org.dbsyncer.sdk.spi.ConnectorService;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.context.ApplicationContext;
+import org.springframework.stereotype.Component;
 import org.springframework.util.Assert;
 
 import javax.annotation.PostConstruct;
@@ -29,10 +33,11 @@ import java.util.concurrent.ConcurrentHashMap;
 /**
  * 连接器工厂
  *
- * @author AE86
- * @version 1.0.0
- * @date 2019/9/18 23:30
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2019-09-18 23:30
  */
+@Component
 public class ConnectorFactory implements DisposableBean {
 
     private final Map<String, ConnectorInstance> pool = new ConcurrentHashMap<>();

dbsyncer-connector/src/main/java/org/dbsyncer/connector/enums/FilterEnum.java

@@ -1,114 +0,0 @@
-package org.dbsyncer.connector.enums;
-
-import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.common.util.NumberUtil;
-import org.dbsyncer.connector.CompareFilter;
-import org.dbsyncer.connector.ConnectorException;
-
-/**
- * 运算符表达式类型
- *
- * @author AE86
- * @version 1.0.0
- * @date 2019/9/26 23:21
- */
-public enum FilterEnum {
-
-    /**
-     * 等于
-     */
-    EQUAL("=", (value, filterValue) -> StringUtil.equals(value, filterValue)),
-    /**
-     * 不等于
-     */
-    NOT_EQUAL("!=", (value, filterValue) -> !StringUtil.equals(value, filterValue)),
-    /**
-     * 大于
-     */
-    GT(">", (value, filterValue) -> NumberUtil.toLong(value) > NumberUtil.toLong(filterValue)),
-    /**
-     * 小于
-     */
-    LT("<", (value, filterValue) -> NumberUtil.toLong(value) < NumberUtil.toLong(filterValue)),
-    /**
-     * 大于等于
-     */
-    GT_AND_EQUAL(">=", (value, filterValue) -> NumberUtil.toLong(value) >= NumberUtil.toLong(filterValue)),
-    /**
-     * 小于等于
-     */
-    LT_AND_EQUAL("<=", (value, filterValue) -> NumberUtil.toLong(value) <= NumberUtil.toLong(filterValue)),
-    /**
-     * 模糊匹配
-     */
-    LIKE("like", (value, filterValue) -> {
-        boolean startsWith = StringUtil.startsWith(filterValue, "%") || StringUtil.startsWith(filterValue, "*");
-        boolean endsWith = StringUtil.endsWith(filterValue, "%") || StringUtil.endsWith(filterValue, "*");
-        String compareValue = StringUtil.replace(filterValue, "%", "");
-        compareValue = StringUtil.replace(compareValue, "*", "");
-        // 模糊匹配
-        if(startsWith && endsWith){
-            return StringUtil.contains(value, compareValue);
-        }
-        // 前缀匹配
-        if(endsWith){
-            return StringUtil.startsWith(value, compareValue);
-        }
-        // 后缀匹配
-        if(startsWith){
-            return StringUtil.endsWith(value, compareValue);
-        }
-        return false;
-    });
-
-    // 运算符名称
-    private String name;
-    // 比较器
-    private CompareFilter compareFilter;
-
-    FilterEnum(String name, CompareFilter compareFilter) {
-        this.name = name;
-        this.compareFilter = compareFilter;
-    }
-
-    /**
-     * 获取表达式
-     *
-     * @param name
-     * @return
-     * @throws ConnectorException
-     */
-    public static FilterEnum getFilterEnum(String name) throws ConnectorException {
-        for (FilterEnum e : FilterEnum.values()) {
-            if (StringUtil.equals(name, e.getName())) {
-                return e;
-            }
-        }
-        throw new ConnectorException(String.format("FilterEnum name \"%s\" does not exist.", name));
-    }
-
-    /**
-     * 获取比较器
-     *
-     * @param filterName
-     * @return
-     * @throws ConnectorException
-     */
-    public static CompareFilter getCompareFilter(String filterName) throws ConnectorException {
-        for (FilterEnum e : FilterEnum.values()) {
-            if (StringUtil.equals(filterName, e.getName())) {
-                return e.getCompareFilter();
-            }
-        }
-        throw new ConnectorException(String.format("FilterEnum name \"%s\" does not exist.", filterName));
-    }
-
-    public String getName() {
-        return name;
-    }
-
-    public CompareFilter getCompareFilter() {
-        return compareFilter;
-    }
-
-}

dbsyncer-connector/src/main/java/org/dbsyncer/connector/file/FileConnector.java

@@ -1,3 +1,6 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.connector.file;
 
 import org.apache.commons.io.IOUtils;
@@ -11,6 +14,7 @@ import org.dbsyncer.sdk.config.CommandConfig;
 import org.dbsyncer.sdk.config.ReaderConfig;
 import org.dbsyncer.sdk.config.WriterBatchConfig;
 import org.dbsyncer.sdk.connector.AbstractConnector;
+import org.dbsyncer.sdk.connector.ConfigValidator;
 import org.dbsyncer.sdk.connector.ConnectorInstance;
 import org.dbsyncer.sdk.enums.ListenerTypeEnum;
 import org.dbsyncer.sdk.listener.Listener;
@@ -38,9 +42,11 @@ import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 
 /**
- * @author AE86
- * @version 1.0.0
- * @date 2022/5/5 23:19
+ * 文件连接器实现
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2022-05-05 23:19
  */
 @Component
 public final class FileConnector extends AbstractConnector implements ConnectorService<FileConnectorInstance, FileConfig> {
@@ -77,6 +83,11 @@ public final class FileConnector extends AbstractConnector implements ConnectorS
         return new FileConnectorInstance(config);
     }
 
+    @Override
+    public ConfigValidator getConfigValidator() {
+        return null;
+    }
+
     @Override
     public void disconnect(FileConnectorInstance connectorInstance) {
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/kafka/KafkaConnector.java

@@ -1,14 +1,18 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.connector.kafka;
 
 import org.dbsyncer.common.model.Result;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.JsonUtil;
+import org.dbsyncer.connector.ConnectorException;
+import org.dbsyncer.connector.config.KafkaConfig;
 import org.dbsyncer.sdk.config.CommandConfig;
 import org.dbsyncer.sdk.config.ReaderConfig;
 import org.dbsyncer.sdk.config.WriterBatchConfig;
 import org.dbsyncer.sdk.connector.AbstractConnector;
-import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.config.KafkaConfig;
+import org.dbsyncer.sdk.connector.ConfigValidator;
 import org.dbsyncer.sdk.connector.ConnectorInstance;
 import org.dbsyncer.sdk.listener.Listener;
 import org.dbsyncer.sdk.model.Field;
@@ -25,6 +29,13 @@ import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
+/**
+ * Kafka连接器实现
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2021-11-22 23:55
+ */
 @Component
 public class KafkaConnector extends AbstractConnector implements ConnectorService<KafkaConnectorInstance, KafkaConfig> {
 
@@ -57,6 +68,11 @@ public class KafkaConnector extends AbstractConnector implements ConnectorServic
         return new KafkaConnectorInstance(config);
     }
 
+    @Override
+    public ConfigValidator getConfigValidator() {
+        return null;
+    }
+
     @Override
     public void disconnect(KafkaConnectorInstance connectorInstance) {
         connectorInstance.close();

dbsyncer-connector/src/main/resources/META-INF/spring.factories

@@ -1,3 +0,0 @@
-# 在这里开启自动状态，遵循spring 扫描规则
-org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
-org.dbsyncer.connector.ConnectorConfiguration

dbsyncer-parser/src/main/java/org/dbsyncer/parser/ProfileComponent.java

@@ -3,7 +3,7 @@
  */
 package org.dbsyncer.parser;
 
-import org.dbsyncer.connector.enums.FilterEnum;
+import org.dbsyncer.sdk.enums.FilterEnum;
 import org.dbsyncer.sdk.enums.QuartzFilterEnum;
 import org.dbsyncer.parser.enums.ConvertEnum;
 import org.dbsyncer.parser.model.ConfigModel;

dbsyncer-parser/src/main/java/org/dbsyncer/parser/impl/ProfileComponentImpl.java

@@ -6,7 +6,7 @@ package org.dbsyncer.parser.impl;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.JsonUtil;
 import org.dbsyncer.connector.ConnectorFactory;
-import org.dbsyncer.connector.enums.FilterEnum;
+import org.dbsyncer.sdk.enums.FilterEnum;
 import org.dbsyncer.sdk.enums.QuartzFilterEnum;
 import org.dbsyncer.parser.ProfileComponent;
 import org.dbsyncer.parser.enums.CommandEnum;

dbsyncer-parser/src/main/java/org/dbsyncer/parser/model/FieldPicker.java

@@ -6,8 +6,8 @@ package org.dbsyncer.parser.model;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.DateFormatUtil;
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.CompareFilter;
-import org.dbsyncer.connector.enums.FilterEnum;
+import org.dbsyncer.sdk.filter.CompareFilter;
+import org.dbsyncer.sdk.enums.FilterEnum;
 import org.dbsyncer.sdk.enums.OperationEnum;
 import org.dbsyncer.sdk.model.Field;
 import org.dbsyncer.sdk.model.Filter;

dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/connector/ConfigValidator.java

@@ -0,0 +1,26 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.sdk.connector;
+
+import java.util.Map;
+
+/**
+ * 连接器配置校验器
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2020-01-07 23:17
+ */
+public interface ConfigValidator<C> {
+
+    /**
+     * 修改配置
+     *
+     * @param connectorConfig
+     * @param params
+     * @return
+     */
+    void modify(C connectorConfig, Map<String, String> params);
+
+}

dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/connector/database/AbstractDatabaseConnector.java

@@ -1,3 +1,6 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.sdk.connector.database;
 
 import org.dbsyncer.common.model.Result;
@@ -11,6 +14,7 @@ import org.dbsyncer.sdk.config.ReaderConfig;
 import org.dbsyncer.sdk.config.SqlBuilderConfig;
 import org.dbsyncer.sdk.config.WriterBatchConfig;
 import org.dbsyncer.sdk.connector.AbstractConnector;
+import org.dbsyncer.sdk.connector.ConfigValidator;
 import org.dbsyncer.sdk.connector.ConnectorInstance;
 import org.dbsyncer.sdk.connector.database.ds.SimpleConnection;
 import org.dbsyncer.sdk.constant.ConnectorConstant;
@@ -48,6 +52,13 @@ import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
+/**
+ * 关系型数据库连接器实现
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2020-01-08 15:17
+ */
 public abstract class AbstractDatabaseConnector extends AbstractConnector implements ConnectorService<DatabaseConnectorInstance, DatabaseConfig>, Database {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
@@ -77,6 +88,11 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector implem
         return new DatabaseConnectorInstance(config);
     }
 
+    @Override
+    public ConfigValidator getConfigValidator() {
+        return null;
+    }
+
     @Override
     public void disconnect(DatabaseConnectorInstance connectorInstance) {
         connectorInstance.close();

dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/enums/FilterEnum.java

@@ -1,16 +1,19 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.sdk.enums;
 
 import org.dbsyncer.common.util.NumberUtil;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.sdk.SdkException;
-import org.dbsyncer.sdk.connector.CompareFilter;
+import org.dbsyncer.sdk.filter.CompareFilter;
 
 /**
  * 运算符表达式类型
  *
- * @author AE86
- * @version 1.0.0
- * @date 2019/9/26 23:21
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2019-09-26 23:21
  */
 public enum FilterEnum {
 
@@ -47,15 +50,15 @@ public enum FilterEnum {
         String compareValue = StringUtil.replace(filterValue, "%", "");
         compareValue = StringUtil.replace(compareValue, "*", "");
         // 模糊匹配
-        if(startsWith && endsWith){
+        if (startsWith && endsWith) {
             return StringUtil.contains(value, compareValue);
         }
         // 前缀匹配
-        if(endsWith){
+        if (endsWith) {
             return StringUtil.startsWith(value, compareValue);
         }
         // 后缀匹配
-        if(startsWith){
+        if (startsWith) {
             return StringUtil.endsWith(value, compareValue);
         }
         return false;

dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/filter/CompareFilter.java

@@ -0,0 +1,17 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.sdk.filter;
+
+/**
+ * 值比较器实现
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2019-11-17 23:56
+ */
+public interface CompareFilter {
+
+    boolean compare(String value, String filterValue);
+
+}

dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/spi/ConnectorService.java

@@ -1,3 +1,6 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.sdk.spi;
 
 import org.dbsyncer.common.model.Result;
@@ -6,6 +9,7 @@ import org.dbsyncer.sdk.config.CommandConfig;
 import org.dbsyncer.sdk.config.DDLConfig;
 import org.dbsyncer.sdk.config.ReaderConfig;
 import org.dbsyncer.sdk.config.WriterBatchConfig;
+import org.dbsyncer.sdk.connector.ConfigValidator;
 import org.dbsyncer.sdk.connector.ConnectorInstance;
 import org.dbsyncer.sdk.enums.ListenerTypeEnum;
 import org.dbsyncer.sdk.listener.Listener;
@@ -21,8 +25,8 @@ import java.util.Map;
  *
  * @param <I> ConnectorInstance
  * @param <C> ConnectorConfig
- * @Version 1.0.0
  * @Author AE86
+ * @Version 1.0.0
  * @Date 2023-11-19 23:24
  */
 public interface ConnectorService<I extends ConnectorInstance, C extends ConnectorConfig> {
@@ -61,6 +65,13 @@ public interface ConnectorService<I extends ConnectorInstance, C extends Connect
      */
     ConnectorInstance connect(C connectorConfig);
 
+    /**
+     * 连接器配置校验器
+     *
+     * @return
+     */
+    ConfigValidator getConfigValidator();
+
     /**
      * 断开连接
      *

dbsyncer-storage/src/main/java/org/dbsyncer/storage/impl/DiskStorageServiceImpl.java

@@ -1,3 +1,6 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.storage.impl;
 
 import org.apache.commons.io.FileUtils;
@@ -13,7 +16,7 @@ import org.apache.lucene.search.highlight.QueryScorer;
 import org.apache.lucene.search.highlight.SimpleHTMLFormatter;
 import org.dbsyncer.common.model.Paging;
 import org.dbsyncer.common.util.CollectionUtils;
-import org.dbsyncer.connector.enums.FilterEnum;
+import org.dbsyncer.sdk.enums.FilterEnum;
 import org.dbsyncer.sdk.enums.OperationEnum;
 import org.dbsyncer.storage.AbstractStorageService;
 import org.dbsyncer.storage.StorageException;
@@ -39,9 +42,9 @@ import java.util.concurrent.ConcurrentHashMap;
 /**
  * 将数据存储在磁盘，基于lucene实现
  *
- * @author AE86
- * @version 1.0.0
- * @date 2019/9/10 23:22
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2023-09-10 23:22
  */
 public class DiskStorageServiceImpl extends AbstractStorageService {
 

dbsyncer-storage/src/main/java/org/dbsyncer/storage/impl/MySQLStorageServiceImpl.java

@@ -1,3 +1,6 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.storage.impl;
 
 import org.apache.commons.io.IOUtils;
@@ -10,7 +13,7 @@ import org.dbsyncer.sdk.config.SqlBuilderConfig;
 import org.dbsyncer.sdk.constant.DatabaseConstant;
 import org.dbsyncer.sdk.connector.database.Database;
 import org.dbsyncer.sdk.connector.database.DatabaseConnectorInstance;
-import org.dbsyncer.connector.enums.FilterEnum;
+import org.dbsyncer.sdk.enums.FilterEnum;
 import org.dbsyncer.sdk.enums.SqlBuilderEnum;
 import org.dbsyncer.sdk.model.Field;
 import org.dbsyncer.sdk.util.DatabaseUtil;
@@ -50,9 +53,9 @@ import java.util.stream.Stream;
 /**
  * 将数据存储在mysql
  *
- * @author AE86
- * @version 1.0.0
- * @date 2019/9/10 23:22
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2020-01-08 15:17
  */
 @Component
 @ConditionalOnProperty(value = "dbsyncer.storage.support.mysql.enabled", havingValue = "true")

dbsyncer-storage/src/main/java/org/dbsyncer/storage/query/AbstractFilter.java

@@ -1,13 +1,18 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.storage.query;
 
 import org.apache.lucene.search.Query;
-import org.dbsyncer.connector.enums.FilterEnum;
+import org.dbsyncer.sdk.enums.FilterEnum;
 import org.dbsyncer.sdk.model.Filter;
 
 /**
- * @author AE86
- * @version 1.0.0
- * @date 2019/11/17 23:56
+ * 过滤语法实现
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2019-11-17 23:56
  */
 public abstract class AbstractFilter extends Filter {
     private boolean enableHighLightSearch;

dbsyncer-storage/src/main/java/org/dbsyncer/storage/query/Query.java

@@ -1,6 +1,9 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.storage.query;
 
-import org.dbsyncer.connector.enums.FilterEnum;
+import org.dbsyncer.sdk.enums.FilterEnum;
 import org.dbsyncer.storage.enums.BinlogSortEnum;
 import org.dbsyncer.storage.enums.IndexFieldResolverEnum;
 import org.dbsyncer.storage.enums.StorageEnum;
@@ -11,9 +14,9 @@ import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
 /**
- * @author AE86
- * @version 1.0.0
- * @date 2019/11/16 22:56
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2020-01-08 15:17
  */
 public class Query {
 

dbsyncer-storage/src/main/java/org/dbsyncer/storage/query/filter/IntFilter.java

@@ -1,9 +1,12 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.storage.query.filter;
 
 import org.apache.lucene.document.IntPoint;
 import org.apache.lucene.search.Query;
 import org.dbsyncer.common.util.NumberUtil;
-import org.dbsyncer.connector.enums.FilterEnum;
+import org.dbsyncer.sdk.enums.FilterEnum;
 import org.dbsyncer.storage.query.AbstractFilter;
 
 public class IntFilter extends AbstractFilter {

dbsyncer-storage/src/main/java/org/dbsyncer/storage/query/filter/LongFilter.java

@@ -1,9 +1,12 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.storage.query.filter;
 
 import org.apache.lucene.document.LongPoint;
 import org.apache.lucene.search.Query;
 import org.dbsyncer.common.util.NumberUtil;
-import org.dbsyncer.connector.enums.FilterEnum;
+import org.dbsyncer.sdk.enums.FilterEnum;
 import org.dbsyncer.storage.query.AbstractFilter;
 
 public class LongFilter extends AbstractFilter {

dbsyncer-storage/src/main/java/org/dbsyncer/storage/query/filter/StringFilter.java

@@ -1,9 +1,12 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.storage.query.filter;
 
 import org.apache.lucene.index.Term;
 import org.apache.lucene.search.Query;
 import org.apache.lucene.search.TermQuery;
-import org.dbsyncer.connector.enums.FilterEnum;
+import org.dbsyncer.sdk.enums.FilterEnum;
 import org.dbsyncer.storage.StorageException;
 import org.dbsyncer.storage.query.AbstractFilter;
 

pom.xml

@@ -11,7 +11,6 @@
     <name>dbsyncer</name>
     <url>https://gitee.com/ghi/dbsyncer</url>
 
-    <!-- 管理的子模块(11个) -->
     <modules>
         <module>dbsyncer-biz</module>
         <module>dbsyncer-cluster</module>