Merge remote-tracking branch 'remotes/origin/master' into V_1.0.0_RC

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-biz/src/main/java/org/dbsyncer/biz/checker/impl/mapping/MappingChecker.java

@@ -3,11 +3,7 @@ package org.dbsyncer.biz.checker.impl.mapping;
 import org.dbsyncer.biz.checker.AbstractChecker;
 import org.dbsyncer.biz.checker.MappingConfigChecker;
 import org.dbsyncer.biz.checker.impl.tablegroup.TableGroupChecker;
-import org.dbsyncer.common.util.BooleanUtil;
-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.common.util.*;
 import org.dbsyncer.listener.config.ListenerConfig;
 import org.dbsyncer.listener.enums.ListenerTypeEnum;
 import org.dbsyncer.manager.Manager;
@@ -138,7 +134,7 @@ public class MappingChecker extends AbstractChecker {
         if (!CollectionUtils.isEmpty(groupAll)) {
             // 手动排序
             String[] sortedTableGroupIds = StringUtil.split(params.get("sortedTableGroupIds"), "|");
-            if(null != sortedTableGroupIds){
+            if (null != sortedTableGroupIds && sortedTableGroupIds.length > 0) {
                 Map<String, TableGroup> tableGroupMap = groupAll.stream().collect(Collectors.toMap(TableGroup::getId, f -> f, (k1, k2) -> k1));
                 groupAll.clear();
                 int size = sortedTableGroupIds.length;

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/AbstractConvertContext.java

@@ -0,0 +1,39 @@
+package org.dbsyncer.common.model;
+
+import org.dbsyncer.common.spi.ConnectorMapper;
+import org.dbsyncer.common.spi.ProxyApplicationContext;
+
+/**
+ * @author AE86
+ * @version 1.0.0
+ * @date 2022/6/30 16:00
+ */
+public abstract class AbstractConvertContext {
+
+    /**
+     * Spring上下文
+     */
+    protected ProxyApplicationContext context;
+
+    /**
+     * 目标源连接实例
+     */
+    protected ConnectorMapper targetConnectorMapper;
+
+    /**
+     * 目标表
+     */
+    protected String targetTableName;
+
+    public ProxyApplicationContext getContext() {
+        return context;
+    }
+
+    public String getTargetTableName() {
+        return targetTableName;
+    }
+
+    public ConnectorMapper getTargetConnectorMapper() {
+        return targetConnectorMapper;
+    }
+}

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

@@ -1,21 +0,0 @@
-package org.dbsyncer.common.model;
-
-import org.dbsyncer.common.spi.ProxyApplicationContext;
-
-/**
- * @author AE86
- * @version 1.0.0
- * @date 2022/6/30 16:00
- */
-public class ConvertContext {
-
-    /**
-     * Spring上下文
-     */
-    protected ProxyApplicationContext context;
-
-    public ProxyApplicationContext getContext() {
-        return context;
-    }
-
-}

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,10 @@ public class FullConvertContext extends ConvertContext {
      */
     private List<Map> targetList;
 
-    public FullConvertContext(ProxyApplicationContext context, 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,10 @@ public class IncrementConvertContext extends ConvertContext {
      */
     private Map target;
 
-    public IncrementConvertContext(ProxyApplicationContext context, 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;
         this.target = target;

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-common/src/main/java/org/dbsyncer/common/spi/ConvertService.java

@@ -27,6 +27,16 @@ public interface ConvertService {
      */
     void convert(IncrementConvertContext context);
 
+    /**
+     * 数据插入后处理接口
+     *
+     * @param context 上下文
+     * @author wangxiri
+     * @date 2022/10/25
+     */
+    default void postProcessAfter(IncrementConvertContext context) {
+    }
+
     /**
      * 版本号
      *

dbsyncer-connector/pom.xml

@@ -35,7 +35,11 @@
             <groupId>com.oracle</groupId>
             <artifactId>ojdbc6</artifactId>
         </dependency>
-
+        <dependency>
+            <groupId>net.postgis</groupId>
+            <artifactId>postgis-jdbc</artifactId>
+            <version>2.5.1</version>
+        </dependency>
         <dependency>
             <groupId>oracle</groupId>
             <artifactId>sdoapi</artifactId>

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/database/DatabaseValueMapper.java

@@ -1,31 +0,0 @@
-package org.dbsyncer.connector.database;
-
-import com.microsoft.sqlserver.jdbc.Geometry;
-import oracle.jdbc.OracleConnection;
-import oracle.spatial.geometry.JGeometry;
-import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.database.ds.SimpleConnection;
-
-import java.sql.SQLException;
-import java.sql.Struct;
-
-public class DatabaseValueMapper {
-
-    private SimpleConnection connection;
-
-    public DatabaseValueMapper(SimpleConnection connection) {
-        this.connection = connection;
-    }
-
-    public Struct getStruct(byte[] val) throws SQLException {
-        if (connection.getConnection() instanceof OracleConnection) {
-            OracleConnection conn = connection.unwrap(OracleConnection.class);
-            Geometry geometry = Geometry.deserialize(val);
-            Double x = geometry.getX();
-            Double y = geometry.getY();
-            JGeometry jGeometry = new JGeometry(x, y, 0);
-            return JGeometry.store(jGeometry, conn);
-        }
-        throw new ConnectorException(String.format("%s can not get STRUCT [%s], val [%s]", getClass().getSimpleName(), val.getClass(), val));
-    }
-}

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

@@ -1,9 +1,11 @@
 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.DatabaseValueMapper;
 import org.dbsyncer.connector.database.ds.SimpleConnection;
 
 import java.sql.Connection;
@@ -27,8 +29,15 @@ public class OtherValueMapper extends AbstractValueMapper<Struct> {
         if (val instanceof byte[]) {
             Object connection = connectorMapper.getConnection();
             if (connection instanceof Connection) {
-                final DatabaseValueMapper mapper = new DatabaseValueMapper((SimpleConnection) connection);
-                return mapper.getStruct((byte[]) val);
+                SimpleConnection simpleConnection = (SimpleConnection) connection;
+                if (simpleConnection instanceof OracleConnection) {
+                    OracleConnection conn = simpleConnection.unwrap(OracleConnection.class);
+                    Geometry geometry = Geometry.deserialize((byte[]) val);
+                    Double x = geometry.getX();
+                    Double y = geometry.getY();
+                    JGeometry jGeometry = new JGeometry(x, y, 0);
+                    return JGeometry.store(jGeometry, conn);
+                }
             }
         }
         throw new ConnectorException(String.format("%s can not find type [%s], val [%s]", getClass().getSimpleName(), val.getClass(), val));

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(), 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;
@@ -58,7 +59,7 @@ public class WriterBufferActuator extends AbstractBufferActuator<WriterRequest,
     @Override
     protected void partition(WriterRequest request, WriterResponse response) {
         response.getDataList().add(request.getRow());
-        if(StringUtil.isNotBlank(request.getMessageId())){
+        if (StringUtil.isNotBlank(request.getMessageId())) {
             response.getMessageIds().add(request.getMessageId());
         }
         if (response.isMerged()) {
@@ -86,17 +87,21 @@ public class WriterBufferActuator extends AbstractBufferActuator<WriterRequest,
         ConvertUtil.convert(tableGroup.getConvert(), targetDataList);
 
         // 4、插件转换
-        pluginFactory.convert(tableGroup.getPlugin(), 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);
 
         // 6、持久化同步结果
         flushStrategy.flushIncrementData(mapping.getMetaId(), result, event);
 
-        // 7、完成处理
+        // 7、执行批量处理后的
+        pluginFactory.postProcessAfter(targetConnectorMapper, plugin, targetTableName, event, sourceDataList, targetDataList);
+
+        // 8、完成处理
         parserStrategy.complete(response.getMessageIds());
     }
 
@@ -112,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,41 @@ public class PluginFactory {
         return Collections.unmodifiableList(plugins);
     }
 
-    public void convert(Plugin plugin, 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, sourceList, targetList));
+            service.get(plugin.getClassName()).convert(new FullConvertContext(applicationContextProxy, targetConnectorMapper, targetTableName, sourceList, targetList));
         }
     }
 
-    public void convert(Plugin plugin, 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()){
+            if (size == targetList.size()) {
                 for (int i = 0; i < size; i++) {
-                    convertService.convert(new IncrementConvertContext(applicationContextProxy, event, sourceList.get(i), targetList.get(i)));
+                    convertService.convert(new IncrementConvertContext(applicationContextProxy, targetConnectorMapper, targetTableName, event, sourceList.get(i), targetList.get(i)));
+                }
+            }
+        }
+    }
+
+    /**
+     * 完成同步后执行处理
+     *
+     * @param targetConnectorMapper
+     * @param plugin
+     * @param targetTableName
+     * @param event
+     * @param sourceList
+     * @param 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, 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;