插件支持获取目标源连接实例 & 捕获同步完成事件

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

@@ -3,9 +3,9 @@ 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.common.model.AbstractConnectorConfig;
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.ConnectorMapper;
-import org.dbsyncer.connector.config.ConnectorConfig;
 import org.dbsyncer.connector.enums.ConnectorEnum;
 import org.dbsyncer.connector.model.Table;
 import org.dbsyncer.manager.Manager;
@@ -53,7 +53,7 @@ public class ConnectorChecker extends AbstractChecker {
         Connector connector = new Connector();
         connector.setName(name);
         connector.setType(ConfigConstant.CONNECTOR);
-        ConnectorConfig config = getConfig(connectorType);
+        AbstractConnectorConfig config = getConfig(connectorType);
         connector.setConfig(config);
 
         // 配置连接器配置
@@ -83,7 +83,7 @@ public class ConnectorChecker extends AbstractChecker {
         this.modifyConfigModel(connector, params);
 
         // 配置连接器配置
-        ConnectorConfig config = connector.getConfig();
+        AbstractConnectorConfig config = connector.getConfig();
         String type = StringUtil.toLowerCaseFirstOne(config.getConnectorType()).concat("ConfigChecker");
         ConnectorConfigChecker checker = map.get(type);
         Assert.notNull(checker, "Checker can not be null.");
@@ -95,11 +95,11 @@ public class ConnectorChecker extends AbstractChecker {
         return connector;
     }
 
-    private ConnectorConfig getConfig(String connectorType) {
-        Class<ConnectorConfig> configClass = (Class<ConnectorConfig>) ConnectorEnum.getConfigClass(connectorType);
+    private AbstractConnectorConfig getConfig(String connectorType) {
+        Class<AbstractConnectorConfig> configClass = (Class<AbstractConnectorConfig>) ConnectorEnum.getConfigClass(connectorType);
         Assert.notNull(configClass, String.format("不支持该连接器类型:%s", connectorType));
         try {
-            ConnectorConfig config = configClass.newInstance();
+            AbstractConnectorConfig config = configClass.newInstance();
             config.setConnectorType(connectorType);
             return config;
         } catch (Exception e) {

dbsyncer-connector/src/main/java/org/dbsyncer/connector/config/ConnectorConfig.java → dbsyncer-common/src/main/java/org/dbsyncer/common/model/AbstractConnectorConfig.java

@@ -1,6 +1,4 @@
-package org.dbsyncer.connector.config;
-
-import org.dbsyncer.connector.enums.ConnectorEnum;
+package org.dbsyncer.common.model;
 
 /**
  * 连接器配置
@@ -8,12 +6,10 @@ import org.dbsyncer.connector.enums.ConnectorEnum;
  * @version 1.0.0
  * @date 2019/9/21 0:01
  */
-public abstract class ConnectorConfig {
+public abstract class AbstractConnectorConfig {
 
     /**
      * 连接器类型
-     *
-     * @see ConnectorEnum
      */
     private String connectorType;
 
@@ -21,7 +17,7 @@ public abstract class ConnectorConfig {
         return connectorType;
     }
 
-    public ConnectorConfig setConnectorType(String connectorType) {
+    public AbstractConnectorConfig setConnectorType(String connectorType) {
         this.connectorType = connectorType;
         return this;
     }

dbsyncer-common/src/main/java/org/dbsyncer/common/model/ConvertContext.java → dbsyncer-common/src/main/java/org/dbsyncer/common/model/AbstractConvertContext.java

@@ -1,5 +1,6 @@
 package org.dbsyncer.common.model;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.common.spi.ProxyApplicationContext;
 
 /**
@@ -7,13 +8,18 @@ import org.dbsyncer.common.spi.ProxyApplicationContext;
  * @version 1.0.0
  * @date 2022/6/30 16:00
  */
-public class ConvertContext {
+public abstract class AbstractConvertContext {
 
     /**
      * Spring上下文
      */
     protected ProxyApplicationContext context;
 
+    /**
+     * 目标源连接实例
+     */
+    protected ConnectorMapper targetConnectorMapper;
+
     /**
      * 目标表
      */
@@ -26,4 +32,8 @@ public class ConvertContext {
     public String getTargetTableName() {
         return targetTableName;
     }
+
+    public ConnectorMapper getTargetConnectorMapper() {
+        return targetConnectorMapper;
+    }
 }

dbsyncer-common/src/main/java/org/dbsyncer/common/model/FullConvertContext.java

@@ -1,5 +1,6 @@
 package org.dbsyncer.common.model;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.common.spi.ProxyApplicationContext;
 
 import java.util.List;
@@ -10,7 +11,7 @@ import java.util.Map;
  * @version 1.0.0
  * @date 2022/6/30 16:04
  */
-public class FullConvertContext extends ConvertContext {
+public class FullConvertContext extends AbstractConvertContext {
 
     /**
      * 全量同步，数据源数据集合
@@ -22,8 +23,9 @@ public class FullConvertContext extends ConvertContext {
      */
     private List<Map> targetList;
 
-    public FullConvertContext(ProxyApplicationContext context, String targetTableName, List<Map> sourceList, List<Map> targetList) {
+    public FullConvertContext(ProxyApplicationContext context, ConnectorMapper targetConnectorMapper, String targetTableName, List<Map> sourceList, List<Map> targetList) {
         this.context = context;
+        this.targetConnectorMapper = targetConnectorMapper;
         this.targetTableName = targetTableName;
         this.sourceList = sourceList;
         this.targetList = targetList;

dbsyncer-common/src/main/java/org/dbsyncer/common/model/IncrementConvertContext.java

@@ -1,5 +1,6 @@
 package org.dbsyncer.common.model;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.common.spi.ProxyApplicationContext;
 
 import java.util.Map;
@@ -9,7 +10,7 @@ import java.util.Map;
  * @version 1.0.0
  * @date 2022/6/30 16:06
  */
-public class IncrementConvertContext extends ConvertContext {
+public class IncrementConvertContext extends AbstractConvertContext {
 
     /**
      * 增量同步，事件（INSERT/UPDATE/DELETE）
@@ -26,8 +27,9 @@ public class IncrementConvertContext extends ConvertContext {
      */
     private Map target;
 
-    public IncrementConvertContext(ProxyApplicationContext context, String targetTableName, String event, Map source, Map target) {
+    public IncrementConvertContext(ProxyApplicationContext context, ConnectorMapper targetConnectorMapper, String targetTableName, String event, Map source, Map target) {
         this.context = context;
+        this.targetConnectorMapper = targetConnectorMapper;
         this.targetTableName = targetTableName;
         this.event = event;
         this.source = source;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/ConnectorMapper.java → dbsyncer-common/src/main/java/org/dbsyncer/common/spi/ConnectorMapper.java

@@ -1,67 +1,68 @@
-package org.dbsyncer.connector;
-
-import org.dbsyncer.connector.config.ConnectorConfig;
-
-/**
- * 连接器实例，管理连接生命周期
- *
- * @param <K> 配置
- * @param <V> 实例
- * @author AE86
- * @version 1.0.0
- * @date 2022/3/20 23:00
- */
-public interface ConnectorMapper<K, V> extends Cloneable {
-
-    /**
-     * 获取连接配置
-     *
-     * @return
-     */
-    default ConnectorConfig getOriginalConfig() {
-        return (ConnectorConfig) getConfig();
-    }
-
-    /**
-     * 获取连接器类型
-     *
-     * @return
-     */
-    default String getConnectorType() {
-        return getOriginalConfig().getConnectorType();
-    }
-
-    /**
-     * 获取连接配置
-     *
-     * @return
-     */
-    K getConfig();
-
-    /**
-     * 设置
-     * @param k
-     */
-    void setConfig(K k);
-
-    /**
-     * 获取连接通道实例
-     *
-     * @return
-     * @throws Exception
-     */
-    V getConnection() throws Exception;
-
-    /**
-     * 关闭连接器
-     */
-    void close();
-
-    /**
-     * 浅拷贝连接器
-     *
-     * @return
-     * @throws CloneNotSupportedException
-     */
-    Object clone() throws CloneNotSupportedException;
+package org.dbsyncer.common.spi;
+
+
+import org.dbsyncer.common.model.AbstractConnectorConfig;
+
+/**
+ * 连接器实例，管理连接生命周期
+ *
+ * @param <K> 配置
+ * @param <V> 实例
+ * @author AE86
+ * @version 1.0.0
+ * @date 2022/3/20 23:00
+ */
+public interface ConnectorMapper<K, V> extends Cloneable {
+
+    /**
+     * 获取连接配置
+     *
+     * @return
+     */
+    default AbstractConnectorConfig getOriginalConfig() {
+        return (AbstractConnectorConfig) getConfig();
+    }
+
+    /**
+     * 获取连接器类型
+     *
+     * @return
+     */
+    default String getConnectorType() {
+        return getOriginalConfig().getConnectorType();
+    }
+
+    /**
+     * 获取连接配置
+     *
+     * @return
+     */
+    K getConfig();
+
+    /**
+     * 设置
+     * @param k
+     */
+    void setConfig(K k);
+
+    /**
+     * 获取连接通道实例
+     *
+     * @return
+     * @throws Exception
+     */
+    V getConnection() throws Exception;
+
+    /**
+     * 关闭连接器
+     */
+    void close();
+
+    /**
+     * 浅拷贝连接器
+     *
+     * @return
+     * @throws CloneNotSupportedException
+     */
+    Object clone() throws CloneNotSupportedException;
 }

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

@@ -1,5 +1,6 @@
 package org.dbsyncer.connector;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.config.WriterBatchConfig;

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

@@ -1,5 +1,7 @@
 package org.dbsyncer.connector;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
+
 import java.lang.reflect.ParameterizedType;
 
 /**

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

@@ -1,7 +1,10 @@
 package org.dbsyncer.connector;
 
 import org.dbsyncer.common.model.Result;
-import org.dbsyncer.connector.config.*;
+import org.dbsyncer.common.spi.ConnectorMapper;
+import org.dbsyncer.connector.config.CommandConfig;
+import org.dbsyncer.connector.config.ReaderConfig;
+import org.dbsyncer.connector.config.WriterBatchConfig;
 import org.dbsyncer.connector.model.MetaInfo;
 import org.dbsyncer.connector.model.Table;
 

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

@@ -1,9 +1,10 @@
 package org.dbsyncer.connector;
 
+import org.dbsyncer.common.model.AbstractConnectorConfig;
 import org.dbsyncer.common.model.Result;
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.connector.config.CommandConfig;
-import org.dbsyncer.connector.config.ConnectorConfig;
 import org.dbsyncer.connector.config.ReaderConfig;
 import org.dbsyncer.connector.config.WriterBatchConfig;
 import org.dbsyncer.connector.enums.ConnectorEnum;
@@ -39,7 +40,7 @@ public class ConnectorFactory implements DisposableBean {
      *
      * @param config
      */
-    public ConnectorMapper connect(ConnectorConfig config) {
+    public ConnectorMapper connect(AbstractConnectorConfig config) {
         Assert.notNull(config, "ConnectorConfig can not be null.");
         Connector connector = getConnector(config.getConnectorType());
         String cacheKey = connector.getConnectorMapperCacheKey(config);
@@ -66,7 +67,7 @@ public class ConnectorFactory implements DisposableBean {
      * @param config
      * @return
      */
-    public boolean refresh(ConnectorConfig config) {
+    public boolean refresh(AbstractConnectorConfig config) {
         Assert.notNull(config, "ConnectorConfig can not be null.");
         Connector connector = getConnector(config.getConnectorType());
         String cacheKey = connector.getConnectorMapperCacheKey(config);
@@ -84,7 +85,7 @@ public class ConnectorFactory implements DisposableBean {
      * @param config
      * @return
      */
-    public boolean isAlive(ConnectorConfig config) {
+    public boolean isAlive(AbstractConnectorConfig config) {
         Assert.notNull(config, "ConnectorConfig can not be null.");
         Connector connector = getConnector(config.getConnectorType());
         String cacheKey = connector.getConnectorMapperCacheKey(config);

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

@@ -1,5 +1,7 @@
 package org.dbsyncer.connector;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
+
 /**
  * 值转换器
  *

dbsyncer-connector/src/main/java/org/dbsyncer/connector/config/CommandConfig.java

@@ -1,5 +1,6 @@
 package org.dbsyncer.connector.config;
 
+import org.dbsyncer.common.model.AbstractConnectorConfig;
 import org.dbsyncer.connector.model.Filter;
 import org.dbsyncer.connector.model.Table;
 
@@ -22,13 +23,13 @@ public class CommandConfig {
 
     private List<Filter> filter;
 
-    private ConnectorConfig connectorConfig;
+    private AbstractConnectorConfig connectorConfig;
 
-    public CommandConfig(String type, Table table, Table originalTable, ConnectorConfig connectorConfig) {
+    public CommandConfig(String type, Table table, Table originalTable, AbstractConnectorConfig connectorConfig) {
         this(type, table, originalTable, connectorConfig, null);
     }
 
-    public CommandConfig(String type, Table table, Table originalTable, ConnectorConfig connectorConfig, List<Filter> filter) {
+    public CommandConfig(String type, Table table, Table originalTable, AbstractConnectorConfig connectorConfig, List<Filter> filter) {
         this.type = type;
         this.table = table;
         this.originalTable = originalTable;
@@ -52,7 +53,7 @@ public class CommandConfig {
         return originalTable;
     }
 
-    public ConnectorConfig getConnectorConfig() {
+    public AbstractConnectorConfig getConnectorConfig() {
         return connectorConfig;
     }
 }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/config/DatabaseConfig.java

@@ -1,5 +1,7 @@
 package org.dbsyncer.connector.config;
 
+import org.dbsyncer.common.model.AbstractConnectorConfig;
+
 import java.util.LinkedHashMap;
 import java.util.Map;
 
@@ -9,7 +11,7 @@ import java.util.Map;
  * @Description: 数据库连接配置
  * @date: 2017年7月20日 下午3:40:59
  */
-public class DatabaseConfig extends ConnectorConfig {
+public class DatabaseConfig extends AbstractConnectorConfig {
 
     /**
      * 驱动

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

@@ -1,12 +1,14 @@
 package org.dbsyncer.connector.config;
 
+import org.dbsyncer.common.model.AbstractConnectorConfig;
+
 /**
  * @author AE86
  * @ClassName: ESConfig
  * @Description: ES连接配置
  * @date: 2021年8月23日 下午8:00:00
  */
-public class ESConfig extends ConnectorConfig {
+public class ESConfig extends AbstractConnectorConfig {
 
     /**
      * 集群地址192.168.1.100:9200,192.168.1.200:9200

dbsyncer-connector/src/main/java/org/dbsyncer/connector/config/FileConfig.java

@@ -1,11 +1,13 @@
 package org.dbsyncer.connector.config;
 
+import org.dbsyncer.common.model.AbstractConnectorConfig;
+
 /**
  * @author AE86
  * @version 1.0.0
  * @date 2022/5/5 23:19
  */
-public class FileConfig extends ConnectorConfig {
+public class FileConfig extends AbstractConnectorConfig {
 
     /**
      * 文件目录

dbsyncer-connector/src/main/java/org/dbsyncer/connector/config/KafkaConfig.java

@@ -1,12 +1,14 @@
 package org.dbsyncer.connector.config;
 
+import org.dbsyncer.common.model.AbstractConnectorConfig;
+
 /**
  * @author AE86
  * @ClassName: KafkaConfig
  * @Description: Kafka连接配置
  * @date: 2021年11月4日 下午8:00:00
  */
-public class KafkaConfig extends ConnectorConfig {
+public class KafkaConfig extends AbstractConnectorConfig {
 
     private String bootstrapServers;
     private String topic;

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

@@ -1,12 +1,12 @@
 package org.dbsyncer.connector.database;
 
 import org.dbsyncer.common.model.Result;
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.AbstractConnector;
 import org.dbsyncer.connector.Connector;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 import org.dbsyncer.connector.config.*;
 import org.dbsyncer.connector.constant.ConnectorConstant;
 import org.dbsyncer.connector.database.ds.SimpleConnection;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/database/DatabaseConnectorMapper.java

@@ -1,7 +1,7 @@
 package org.dbsyncer.connector.database;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 import org.dbsyncer.connector.config.DatabaseConfig;
 import org.dbsyncer.connector.database.ds.SimpleDataSource;
 import org.dbsyncer.connector.util.DatabaseUtil;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/es/ESConnector.java

@@ -1,13 +1,13 @@
 package org.dbsyncer.connector.es;
 
 import org.dbsyncer.common.model.Result;
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.JsonUtil;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.AbstractConnector;
 import org.dbsyncer.connector.Connector;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 import org.dbsyncer.connector.config.CommandConfig;
 import org.dbsyncer.connector.config.ESConfig;
 import org.dbsyncer.connector.config.ReaderConfig;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/es/ESConnectorMapper.java

@@ -1,6 +1,6 @@
 package org.dbsyncer.connector.es;
 
-import org.dbsyncer.connector.ConnectorMapper;
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.config.ESConfig;
 import org.dbsyncer.connector.util.ESUtil;
 import org.elasticsearch.client.RestHighLevelClient;

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

@@ -3,12 +3,12 @@ package org.dbsyncer.connector.file;
 import org.apache.commons.io.IOUtils;
 import org.apache.commons.io.LineIterator;
 import org.dbsyncer.common.model.Result;
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.AbstractConnector;
 import org.dbsyncer.connector.Connector;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 import org.dbsyncer.connector.config.CommandConfig;
 import org.dbsyncer.connector.config.FileConfig;
 import org.dbsyncer.connector.config.ReaderConfig;
@@ -24,7 +24,10 @@ import org.springframework.util.Assert;
 import java.io.*;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 

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

@@ -1,7 +1,7 @@
 package org.dbsyncer.connector.file;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.common.util.JsonUtil;
-import org.dbsyncer.connector.ConnectorMapper;
 import org.dbsyncer.connector.config.FileConfig;
 import org.dbsyncer.connector.model.Field;
 import org.dbsyncer.connector.model.FileSchema;

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

@@ -1,13 +1,16 @@
 package org.dbsyncer.connector.kafka;
 
 import org.dbsyncer.common.model.Result;
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.JsonUtil;
 import org.dbsyncer.connector.AbstractConnector;
 import org.dbsyncer.connector.Connector;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
-import org.dbsyncer.connector.config.*;
+import org.dbsyncer.connector.config.CommandConfig;
+import org.dbsyncer.connector.config.KafkaConfig;
+import org.dbsyncer.connector.config.ReaderConfig;
+import org.dbsyncer.connector.config.WriterBatchConfig;
 import org.dbsyncer.connector.model.Field;
 import org.dbsyncer.connector.model.MetaInfo;
 import org.dbsyncer.connector.model.Table;

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

@@ -1,6 +1,6 @@
 package org.dbsyncer.connector.kafka;
 
-import org.dbsyncer.connector.ConnectorMapper;
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.config.KafkaConfig;
 import org.dbsyncer.connector.util.KafkaUtil;
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/BigintValueMapper.java

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 import java.math.BigDecimal;
 import java.math.BigInteger;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/BinaryValueMapper.java

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 /**
  * @author AE86

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/BitValueMapper.java

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 import java.nio.ByteBuffer;
 import java.util.BitSet;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/BlobValueMapper.java

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 import java.sql.Blob;
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/CharValueMapper.java

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 /**
  * @author AE86

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/ClobValueMapper.java

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 import java.sql.Clob;
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/DateValueMapper.java

@@ -1,9 +1,9 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.common.util.DateFormatUtil;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 import java.sql.Date;
 import java.sql.Timestamp;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/DecimalValueMapper.java

@@ -1,9 +1,9 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.common.util.NumberUtil;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 import java.math.BigDecimal;
 import java.math.BigInteger;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/DoubleValueMapper.java

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 import java.math.BigDecimal;
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/FloatValueMapper.java

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 import java.math.BigDecimal;
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/GeometryValueMapper.java

@@ -2,9 +2,9 @@ package org.dbsyncer.connector.schema;
 
 import com.microsoft.sqlserver.jdbc.Geometry;
 import com.microsoft.sqlserver.jdbc.SQLServerException;
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 /**
  * @author AE86

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/IntegerValueMapper.java

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 import java.math.BigDecimal;
 import java.math.BigInteger;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/LongVarBinaryValueMapper.java

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 /**
  * @author AE86

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/LongVarcharValueMapper.java

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 import java.sql.Date;
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/NCharValueMapper.java

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 /**
  * @author AE86

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/NClobValueMapper.java

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 import java.sql.NClob;
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/NVarcharValueMapper.java

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 /**
  * @author AE86

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/NumberValueMapper.java

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 import java.math.BigDecimal;
 import java.math.BigInteger;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/OtherValueMapper.java

@@ -3,9 +3,9 @@ package org.dbsyncer.connector.schema;
 import com.microsoft.sqlserver.jdbc.Geometry;
 import oracle.jdbc.OracleConnection;
 import oracle.spatial.geometry.JGeometry;
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 import org.dbsyncer.connector.database.ds.SimpleConnection;
 
 import java.sql.Connection;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/RealValueMapper.java

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 /**
  * @author AE86

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/RowIdValueMapper.java

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 import java.sql.RowId;
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/SmallintValueMapper.java

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 /**
  * @author AE86

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/TimeValueMapper.java

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 import java.sql.Time;
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/TimestampValueMapper.java

@@ -1,9 +1,9 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.common.util.DateFormatUtil;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 import java.sql.Date;
 import java.sql.Timestamp;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/TinyintValueMapper.java

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 /**
  * @author AE86

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/VarBinaryValueMapper.java

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 /**
  * @author AE86

dbsyncer-connector/src/main/java/org/dbsyncer/connector/schema/VarcharValueMapper.java

@@ -1,9 +1,9 @@
 package org.dbsyncer.connector.schema;
 
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.common.util.DateFormatUtil;
 import org.dbsyncer.connector.AbstractValueMapper;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorMapper;
 
 import java.time.LocalDate;
 import java.time.LocalDateTime;

dbsyncer-listener/src/main/java/org/dbsyncer/listener/AbstractExtractor.java

@@ -2,9 +2,9 @@ package org.dbsyncer.listener;
 
 import org.dbsyncer.common.event.Event;
 import org.dbsyncer.common.event.RowChangedEvent;
+import org.dbsyncer.common.model.AbstractConnectorConfig;
 import org.dbsyncer.common.scheduled.ScheduledTaskService;
 import org.dbsyncer.connector.ConnectorFactory;
-import org.dbsyncer.connector.config.ConnectorConfig;
 import org.dbsyncer.connector.constant.ConnectorConstant;
 import org.dbsyncer.listener.config.ListenerConfig;
 import org.slf4j.Logger;
@@ -27,7 +27,7 @@ public abstract class AbstractExtractor implements Extractor {
     protected String metaId;
     protected ConnectorFactory connectorFactory;
     protected ScheduledTaskService scheduledTaskService;
-    protected ConnectorConfig connectorConfig;
+    protected AbstractConnectorConfig connectorConfig;
     protected ListenerConfig listenerConfig;
     protected Map<String, String> snapshot;
     protected Set<String> filterTable;
@@ -120,7 +120,7 @@ public abstract class AbstractExtractor implements Extractor {
         this.scheduledTaskService = scheduledTaskService;
     }
 
-    public void setConnectorConfig(ConnectorConfig connectorConfig) {
+    public void setConnectorConfig(AbstractConnectorConfig connectorConfig) {
         this.connectorConfig = connectorConfig;
     }
 

dbsyncer-listener/src/main/java/org/dbsyncer/listener/quartz/AbstractQuartzExtractor.java

@@ -3,10 +3,10 @@ package org.dbsyncer.listener.quartz;
 import org.dbsyncer.common.event.RowChangedEvent;
 import org.dbsyncer.common.model.Result;
 import org.dbsyncer.common.scheduled.ScheduledTaskJob;
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.common.util.UUIDUtil;
-import org.dbsyncer.connector.ConnectorMapper;
 import org.dbsyncer.connector.config.ReaderConfig;
 import org.dbsyncer.connector.constant.ConnectorConstant;
 import org.dbsyncer.listener.AbstractExtractor;

dbsyncer-manager/src/main/java/org/dbsyncer/manager/Manager.java

@@ -1,13 +1,13 @@
 package org.dbsyncer.manager;
 
+import org.dbsyncer.common.model.AbstractConnectorConfig;
 import org.dbsyncer.common.model.Paging;
-import org.dbsyncer.connector.ConnectorMapper;
-import org.dbsyncer.connector.config.ConnectorConfig;
-import org.dbsyncer.connector.model.MetaInfo;
-import org.dbsyncer.connector.model.Table;
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.enums.ConnectorEnum;
 import org.dbsyncer.connector.enums.FilterEnum;
 import org.dbsyncer.connector.enums.OperationEnum;
+import org.dbsyncer.connector.model.MetaInfo;
+import org.dbsyncer.connector.model.Table;
 import org.dbsyncer.listener.enums.QuartzFilterEnum;
 import org.dbsyncer.parser.enums.ConvertEnum;
 import org.dbsyncer.parser.model.*;
@@ -39,11 +39,11 @@ public interface Manager extends Executor {
     List<ProjectGroup> getProjectGroupAll();
 
     // Connector
-    ConnectorMapper connect(ConnectorConfig config);
+    ConnectorMapper connect(AbstractConnectorConfig config);
 
-    boolean refreshConnectorConfig(ConnectorConfig config);
+    boolean refreshConnectorConfig(AbstractConnectorConfig config);
 
-    boolean isAliveConnectorConfig(ConnectorConfig config);
+    boolean isAliveConnectorConfig(AbstractConnectorConfig config);
 
     List<Table> getTable(ConnectorMapper config);
 

dbsyncer-manager/src/main/java/org/dbsyncer/manager/ManagerFactory.java

@@ -1,15 +1,15 @@
 package org.dbsyncer.manager;
 
 import org.dbsyncer.common.event.ClosedEvent;
-import org.dbsyncer.common.util.CollectionUtils;
+import org.dbsyncer.common.model.AbstractConnectorConfig;
 import org.dbsyncer.common.model.Paging;
-import org.dbsyncer.connector.ConnectorMapper;
-import org.dbsyncer.connector.config.ConnectorConfig;
-import org.dbsyncer.connector.model.MetaInfo;
-import org.dbsyncer.connector.model.Table;
+import org.dbsyncer.common.spi.ConnectorMapper;
+import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.connector.enums.ConnectorEnum;
 import org.dbsyncer.connector.enums.FilterEnum;
 import org.dbsyncer.connector.enums.OperationEnum;
+import org.dbsyncer.connector.model.MetaInfo;
+import org.dbsyncer.connector.model.Table;
 import org.dbsyncer.listener.enums.QuartzFilterEnum;
 import org.dbsyncer.manager.config.OperationConfig;
 import org.dbsyncer.manager.config.QueryConfig;
@@ -98,17 +98,17 @@ public class ManagerFactory implements Manager, ApplicationListener<ClosedEvent>
     }
 
     @Override
-    public ConnectorMapper connect(ConnectorConfig config) {
+    public ConnectorMapper connect(AbstractConnectorConfig config) {
         return parser.connect(config);
     }
 
     @Override
-    public boolean refreshConnectorConfig(ConnectorConfig config) {
+    public boolean refreshConnectorConfig(AbstractConnectorConfig config) {
         return parser.refreshConnectorConfig(config);
     }
 
     @Override
-    public boolean isAliveConnectorConfig(ConnectorConfig config) {
+    public boolean isAliveConnectorConfig(AbstractConnectorConfig config) {
         return parser.isAliveConnectorConfig(config);
     }
 

dbsyncer-manager/src/main/java/org/dbsyncer/manager/config/ExtractorConfig.java

@@ -1,51 +0,0 @@
-package org.dbsyncer.manager.config;
-
-import org.dbsyncer.common.event.Event;
-import org.dbsyncer.connector.config.ConnectorConfig;
-import org.dbsyncer.listener.config.ListenerConfig;
-
-import java.util.Map;
-
-/**
- * @version 1.0.0
- * @Author AE86
- * @Date 2020-05-25 23:52
- */
-public class ExtractorConfig {
-
-    private ConnectorConfig connectorConfig;
-    private ListenerConfig listenerConfig;
-    private Map<String, String> map;
-    private Event event;
-
-    /**
-     * 抽取器配置
-     *
-     * @param connectorConfig 连接器配置
-     * @param listenerConfig 监听配置
-     * @param map 增量元信息
-     * @param event 监听器
-     */
-    public ExtractorConfig(ConnectorConfig connectorConfig, ListenerConfig listenerConfig, Map<String, String> map, Event event) {
-        this.connectorConfig = connectorConfig;
-        this.listenerConfig = listenerConfig;
-        this.map = map;
-        this.event = event;
-    }
-
-    public ConnectorConfig getConnectorConfig() {
-        return connectorConfig;
-    }
-
-    public ListenerConfig getListenerConfig() {
-        return listenerConfig;
-    }
-
-    public Map<String, String> getMap() {
-        return map;
-    }
-
-    public Event getEvent() {
-        return event;
-    }
-}

dbsyncer-manager/src/main/java/org/dbsyncer/manager/puller/IncrementPuller.java

@@ -2,11 +2,11 @@ package org.dbsyncer.manager.puller;
 
 import org.dbsyncer.common.event.Event;
 import org.dbsyncer.common.event.RowChangedEvent;
+import org.dbsyncer.common.model.AbstractConnectorConfig;
 import org.dbsyncer.common.scheduled.ScheduledTaskJob;
 import org.dbsyncer.common.scheduled.ScheduledTaskService;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.connector.ConnectorFactory;
-import org.dbsyncer.connector.config.ConnectorConfig;
 import org.dbsyncer.connector.model.Field;
 import org.dbsyncer.connector.model.Table;
 import org.dbsyncer.listener.AbstractExtractor;
@@ -132,7 +132,7 @@ public class IncrementPuller extends AbstractPuller implements ScheduledTaskJob
 
     private AbstractExtractor getExtractor(Mapping mapping, Connector connector, List<TableGroup> list, Meta meta)
             throws InstantiationException, IllegalAccessException {
-        ConnectorConfig connectorConfig = connector.getConfig();
+        AbstractConnectorConfig connectorConfig = connector.getConfig();
         ListenerConfig listenerConfig = mapping.getListener();
 
         // timing/log
@@ -163,7 +163,7 @@ public class IncrementPuller extends AbstractPuller implements ScheduledTaskJob
         throw new ManagerException("未知的监听配置.");
     }
 
-    private void setExtractorConfig(AbstractExtractor extractor, ConnectorConfig connector, ListenerConfig listener,
+    private void setExtractorConfig(AbstractExtractor extractor, AbstractConnectorConfig connector, ListenerConfig listener,
                                     Map<String, String> snapshot, AbstractListener event) {
         extractor.setConnectorFactory(connectorFactory);
         extractor.setScheduledTaskService(scheduledTaskService);

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

@@ -1,14 +1,14 @@
 package org.dbsyncer.parser;
 
 import org.dbsyncer.common.event.RowChangedEvent;
+import org.dbsyncer.common.model.AbstractConnectorConfig;
 import org.dbsyncer.common.model.Result;
-import org.dbsyncer.connector.ConnectorMapper;
-import org.dbsyncer.connector.config.ConnectorConfig;
-import org.dbsyncer.connector.model.MetaInfo;
-import org.dbsyncer.connector.model.Table;
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.enums.ConnectorEnum;
 import org.dbsyncer.connector.enums.FilterEnum;
 import org.dbsyncer.connector.enums.OperationEnum;
+import org.dbsyncer.connector.model.MetaInfo;
+import org.dbsyncer.connector.model.Table;
 import org.dbsyncer.listener.enums.QuartzFilterEnum;
 import org.dbsyncer.parser.enums.ConvertEnum;
 import org.dbsyncer.parser.model.*;
@@ -31,14 +31,14 @@ public interface Parser {
      * @param config
      * @return
      */
-    ConnectorMapper connect(ConnectorConfig config);
+    ConnectorMapper connect(AbstractConnectorConfig config);
 
     /**
      * 刷新连接配置
      *
      * @param config
      */
-    boolean refreshConnectorConfig(ConnectorConfig config);
+    boolean refreshConnectorConfig(AbstractConnectorConfig config);
 
     /**
      * 连接配置是否可用
@@ -46,7 +46,7 @@ public interface Parser {
      * @param config
      * @return
      */
-    boolean isAliveConnectorConfig(ConnectorConfig config);
+    boolean isAliveConnectorConfig(AbstractConnectorConfig config);
 
     /**
      * 获取连接器表

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

@@ -4,13 +4,13 @@ import com.alibaba.fastjson.JSONException;
 import com.alibaba.fastjson.JSONObject;
 import org.dbsyncer.cache.CacheService;
 import org.dbsyncer.common.event.RowChangedEvent;
+import org.dbsyncer.common.model.AbstractConnectorConfig;
 import org.dbsyncer.common.model.Result;
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.JsonUtil;
 import org.dbsyncer.connector.ConnectorFactory;
-import org.dbsyncer.connector.ConnectorMapper;
 import org.dbsyncer.connector.config.CommandConfig;
-import org.dbsyncer.connector.config.ConnectorConfig;
 import org.dbsyncer.connector.config.ReaderConfig;
 import org.dbsyncer.connector.config.WriterBatchConfig;
 import org.dbsyncer.connector.constant.ConnectorConstant;
@@ -31,6 +31,7 @@ import org.dbsyncer.parser.strategy.ParserStrategy;
 import org.dbsyncer.parser.util.ConvertUtil;
 import org.dbsyncer.parser.util.PickerUtil;
 import org.dbsyncer.plugin.PluginFactory;
+import org.dbsyncer.plugin.config.Plugin;
 import org.dbsyncer.storage.enums.StorageDataStatusEnum;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -86,17 +87,17 @@ public class ParserFactory implements Parser {
     private ParserStrategy parserStrategy;
 
     @Override
-    public ConnectorMapper connect(ConnectorConfig config) {
+    public ConnectorMapper connect(AbstractConnectorConfig config) {
         return connectorFactory.connect(config);
     }
 
     @Override
-    public boolean refreshConnectorConfig(ConnectorConfig config) {
+    public boolean refreshConnectorConfig(AbstractConnectorConfig config) {
         return connectorFactory.refresh(config);
     }
 
     @Override
-    public boolean isAliveConnectorConfig(ConnectorConfig config) {
+    public boolean isAliveConnectorConfig(AbstractConnectorConfig config) {
         boolean alive = false;
         try {
             alive = connectorFactory.isAlive(config);
@@ -141,8 +142,8 @@ public class ParserFactory implements Parser {
 
     @Override
     public Map<String, String> getCommand(Mapping mapping, TableGroup tableGroup) {
-        ConnectorConfig sConnConfig = getConnectorConfig(mapping.getSourceConnectorId());
-        ConnectorConfig tConnConfig = getConnectorConfig(mapping.getTargetConnectorId());
+        AbstractConnectorConfig sConnConfig = getConnectorConfig(mapping.getSourceConnectorId());
+        AbstractConnectorConfig tConnConfig = getConnectorConfig(mapping.getTargetConnectorId());
         Table sourceTable = tableGroup.getSourceTable();
         Table targetTable = tableGroup.getTargetTable();
         Table sTable = new Table(sourceTable.getName(), sourceTable.getType(), new ArrayList<>());
@@ -180,7 +181,7 @@ public class ParserFactory implements Parser {
             Assert.notNull(connector, "Connector can not be null.");
             String connectorType = config.getString("connectorType");
             Class<?> configClass = ConnectorEnum.getConfigClass(connectorType);
-            ConnectorConfig obj = (ConnectorConfig) JsonUtil.jsonToObj(config.toString(), configClass);
+            AbstractConnectorConfig obj = (AbstractConnectorConfig) JsonUtil.jsonToObj(config.toString(), configClass);
             connector.setConfig(obj);
 
             return connector;
@@ -232,9 +233,9 @@ public class ParserFactory implements Parser {
         final String sourceConnectorId = mapping.getSourceConnectorId();
         final String targetConnectorId = mapping.getTargetConnectorId();
 
-        ConnectorConfig sConfig = getConnectorConfig(sourceConnectorId);
+        AbstractConnectorConfig sConfig = getConnectorConfig(sourceConnectorId);
         Assert.notNull(sConfig, "数据源配置不能为空.");
-        ConnectorConfig tConfig = getConnectorConfig(targetConnectorId);
+        AbstractConnectorConfig tConfig = getConnectorConfig(targetConnectorId);
         Assert.notNull(tConfig, "目标源配置不能为空.");
         TableGroup group = PickerUtil.mergeTableGroupConfig(mapping, tableGroup);
         Map<String, String> command = group.getCommand();
@@ -273,18 +274,22 @@ public class ParserFactory implements Parser {
             ConvertUtil.convert(group.getConvert(), target);
 
             // 4、插件转换
-            pluginFactory.convert(group.getPlugin(), tTableName, data, target);
+            Plugin plugin = group.getPlugin();
+            pluginFactory.convert(tConnectorMapper, plugin, tTableName, data, target);
 
             // 5、写入目标源
             BatchWriter batchWriter = new BatchWriter(tConnectorMapper, command, tTableName, ConnectorConstant.OPERTION_INSERT, picker.getTargetFields(), target, batchSize);
             Result writer = writeBatch(batchWriter, executorService);
 
-            // 6、更新结果
+            // 6、执行批量处理后的
+            pluginFactory.postProcessAfter(tConnectorMapper, plugin, tTableName, ConnectorConstant.OPERTION_INSERT, data, target);
+
+            // 7、更新结果
             task.setPageIndex(task.getPageIndex() + 1);
             task.setCursor(getLastCursor(data, pk));
             flush(task, writer);
 
-            // 7、判断尾页
+            // 8、判断尾页
             if (data.size() < pageSize) {
                 logger.info("完成全量:{}, [{}] >> [{}]", metaId, sTableName, tTableName);
                 break;
@@ -402,7 +407,7 @@ public class ParserFactory implements Parser {
      * @param connectorId
      * @return
      */
-    private ConnectorConfig getConnectorConfig(String connectorId) {
+    private AbstractConnectorConfig getConnectorConfig(String connectorId) {
         return getConnector(connectorId).getConfig();
     }
 

dbsyncer-parser/src/main/java/org/dbsyncer/parser/flush/impl/WriterBufferActuator.java

@@ -2,11 +2,11 @@ package org.dbsyncer.parser.flush.impl;
 
 import org.dbsyncer.cache.CacheService;
 import org.dbsyncer.common.config.BufferActuatorConfig;
+import org.dbsyncer.common.model.AbstractConnectorConfig;
 import org.dbsyncer.common.model.Result;
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.ConnectorFactory;
-import org.dbsyncer.connector.ConnectorMapper;
-import org.dbsyncer.connector.config.ConnectorConfig;
 import org.dbsyncer.parser.ParserFactory;
 import org.dbsyncer.parser.flush.AbstractBufferActuator;
 import org.dbsyncer.parser.model.*;
@@ -14,6 +14,7 @@ import org.dbsyncer.parser.strategy.FlushStrategy;
 import org.dbsyncer.parser.strategy.ParserStrategy;
 import org.dbsyncer.parser.util.ConvertUtil;
 import org.dbsyncer.plugin.PluginFactory;
+import org.dbsyncer.plugin.config.Plugin;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import org.springframework.util.Assert;
@@ -86,10 +87,11 @@ public class WriterBufferActuator extends AbstractBufferActuator<WriterRequest,
         ConvertUtil.convert(tableGroup.getConvert(), targetDataList);
 
         // 4、插件转换
-        pluginFactory.convert(tableGroup.getPlugin(), targetTableName, event, sourceDataList, targetDataList);
+        ConnectorMapper targetConnectorMapper = connectorFactory.connect(getConnectorConfig(mapping.getTargetConnectorId()));
+        Plugin plugin = tableGroup.getPlugin();
+        pluginFactory.convert(targetConnectorMapper, plugin, targetTableName, event, sourceDataList, targetDataList);
 
         // 5、批量执行同步
-        ConnectorMapper targetConnectorMapper = connectorFactory.connect(getConnectorConfig(mapping.getTargetConnectorId()));
         BatchWriter batchWriter = new BatchWriter(targetConnectorMapper, tableGroup.getCommand(), targetTableName, event, picker.getTargetFields(), targetDataList, bufferActuatorConfig.getWriterBatchCount());
         Result result = parserFactory.writeBatch(batchWriter);
 
@@ -97,7 +99,7 @@ public class WriterBufferActuator extends AbstractBufferActuator<WriterRequest,
         flushStrategy.flushIncrementData(mapping.getMetaId(), result, event);
 
         // 7、执行批量处理后的
-        pluginFactory.postProcessAfter(tableGroup.getPlugin(), targetTableName, event, sourceDataList, targetDataList);
+        pluginFactory.postProcessAfter(targetConnectorMapper, plugin, targetTableName, event, sourceDataList, targetDataList);
 
         // 8、完成处理
         parserStrategy.complete(response.getMessageIds());
@@ -115,7 +117,7 @@ public class WriterBufferActuator extends AbstractBufferActuator<WriterRequest,
      * @param connectorId
      * @return
      */
-    private ConnectorConfig getConnectorConfig(String connectorId) {
+    private AbstractConnectorConfig getConnectorConfig(String connectorId) {
         Assert.hasText(connectorId, "Connector id can not be empty.");
         Connector conn = cacheService.get(connectorId, Connector.class);
         Assert.notNull(conn, "Connector can not be null.");

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

@@ -1,6 +1,6 @@
 package org.dbsyncer.parser.model;
 
-import org.dbsyncer.connector.ConnectorMapper;
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.connector.model.Field;
 
 import java.util.List;

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

@@ -1,6 +1,6 @@
 package org.dbsyncer.parser.model;
 
-import org.dbsyncer.connector.config.ConnectorConfig;
+import org.dbsyncer.common.model.AbstractConnectorConfig;
 import org.dbsyncer.connector.model.Table;
 
 import java.util.List;
@@ -20,7 +20,7 @@ public class Connector extends ConfigModel{
     /**
      * 连接器配置
      */
-    private ConnectorConfig config;
+    private AbstractConnectorConfig config;
 
     public List<Table> getTable() {
         return table;
@@ -31,11 +31,11 @@ public class Connector extends ConfigModel{
         return this;
     }
 
-    public ConnectorConfig getConfig() {
+    public AbstractConnectorConfig getConfig() {
         return config;
     }
 
-    public Connector setConfig(ConnectorConfig config) {
+    public Connector setConfig(AbstractConnectorConfig config) {
         this.config = config;
         return this;
     }

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

@@ -1,7 +1,7 @@
 package org.dbsyncer.parser.model;
 
+import org.dbsyncer.common.model.AbstractConnectorConfig;
 import org.dbsyncer.common.util.CollectionUtils;
-import org.dbsyncer.connector.config.ConnectorConfig;
 import org.dbsyncer.connector.model.Field;
 import org.springframework.util.Assert;
 
@@ -57,7 +57,7 @@ public class Picker {
         }
     }
 
-    public String getSourcePrimaryKeyName(ConnectorConfig config) {
+    public String getSourcePrimaryKeyName(AbstractConnectorConfig config) {
         for (Field f : sourceFields) {
             if (null != f && f.isPk()) {
                 return f.getName();

dbsyncer-parser/src/main/test/ConnectorParserTest.java

@@ -1,8 +1,8 @@
 import com.alibaba.fastjson.JSONException;
 import com.alibaba.fastjson.JSONObject;
 import org.apache.commons.io.FileUtils;
+import org.dbsyncer.common.model.AbstractConnectorConfig;
 import org.dbsyncer.common.util.JsonUtil;
-import org.dbsyncer.connector.config.ConnectorConfig;
 import org.dbsyncer.connector.enums.ConnectorEnum;
 import org.dbsyncer.parser.model.Connector;
 import org.dbsyncer.parser.model.Mapping;
@@ -34,7 +34,7 @@ public class ConnectorParserTest {
         String connectorType = config.getString("connectorType");
         Class<?> configClass = ConnectorEnum.getConfigClass(connectorType);
         Object obj = JsonUtil.jsonToObj(config.toString(), configClass);
-        connector.setConfig((ConnectorConfig) obj);
+        connector.setConfig((AbstractConnectorConfig) obj);
         System.out.println(connector);
     }
 

dbsyncer-plugin/pom.xml

@@ -18,5 +18,12 @@
             <version>${project.parent.version}</version>
         </dependency>
 
+        <!-- Connector 连接器 -->
+        <dependency>
+            <groupId>org.ghi</groupId>
+            <artifactId>dbsyncer-connector</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
+
     </dependencies>
 </project>

dbsyncer-plugin/src/main/java/org/dbsyncer/plugin/PluginFactory.java

@@ -3,6 +3,7 @@ package org.dbsyncer.plugin;
 import org.apache.commons.io.FileUtils;
 import org.dbsyncer.common.model.FullConvertContext;
 import org.dbsyncer.common.model.IncrementConvertContext;
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.common.spi.ConvertService;
 import org.dbsyncer.common.spi.ProxyApplicationContext;
 import org.dbsyncer.common.util.CollectionUtils;
@@ -96,19 +97,19 @@ public class PluginFactory {
         return Collections.unmodifiableList(plugins);
     }
 
-    public void convert(Plugin plugin, String targetTableName, List<Map> sourceList, List<Map> targetList) {
+    public void convert(ConnectorMapper targetConnectorMapper, Plugin plugin, String targetTableName, List<Map> sourceList, List<Map> targetList) {
         if (null != plugin && service.containsKey(plugin.getClassName())) {
-            service.get(plugin.getClassName()).convert(new FullConvertContext(applicationContextProxy, targetTableName, sourceList, targetList));
+            service.get(plugin.getClassName()).convert(new FullConvertContext(applicationContextProxy, targetConnectorMapper, targetTableName, sourceList, targetList));
         }
     }
 
-    public void convert(Plugin plugin, String targetTableName, String event, List<Map> sourceList, List<Map> targetList) {
+    public void convert(ConnectorMapper targetConnectorMapper, Plugin plugin,String targetTableName, String event, List<Map> sourceList, List<Map> targetList) {
         if (null != plugin && service.containsKey(plugin.getClassName())) {
             ConvertService convertService = service.get(plugin.getClassName());
             int size = sourceList.size();
             if (size == targetList.size()) {
                 for (int i = 0; i < size; i++) {
-                    convertService.convert(new IncrementConvertContext(applicationContextProxy, targetTableName, event, sourceList.get(i), targetList.get(i)));
+                    convertService.convert(new IncrementConvertContext(applicationContextProxy, targetConnectorMapper, targetTableName, event, sourceList.get(i), targetList.get(i)));
                 }
             }
         }
@@ -117,19 +118,20 @@ public class PluginFactory {
     /**
      * 完成同步后执行处理
      *
+     * @param targetConnectorMapper
      * @param plugin
      * @param targetTableName
      * @param event
      * @param sourceList
      * @param targetList
      */
-    public void postProcessAfter(Plugin plugin, String targetTableName, String event, List<Map> sourceList, List<Map> targetList) {
+    public void postProcessAfter(ConnectorMapper targetConnectorMapper, Plugin plugin, String targetTableName, String event, List<Map> sourceList, List<Map> targetList) {
         if (null != plugin && service.containsKey(plugin.getClassName())) {
             ConvertService convertService = service.get(plugin.getClassName());
             int size = sourceList.size();
             if (size == targetList.size()) {
                 for (int i = 0; i < size; i++) {
-                    convertService.postProcessAfter(new IncrementConvertContext(applicationContextProxy, targetTableName, event, sourceList.get(i), targetList.get(i)));
+                    convertService.postProcessAfter(new IncrementConvertContext(applicationContextProxy, targetConnectorMapper, targetTableName, event, sourceList.get(i), targetList.get(i)));
                 }
             }
         }

dbsyncer-storage/src/main/java/org/dbsyncer/storage/support/MysqlStorageServiceImpl.java

@@ -1,10 +1,10 @@
 package org.dbsyncer.storage.support;
 
 import org.dbsyncer.common.model.Paging;
+import org.dbsyncer.common.spi.ConnectorMapper;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.ConnectorFactory;
-import org.dbsyncer.connector.ConnectorMapper;
 import org.dbsyncer.connector.config.DatabaseConfig;
 import org.dbsyncer.connector.config.SqlBuilderConfig;
 import org.dbsyncer.connector.config.WriterBatchConfig;