fix bug

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

@@ -3,7 +3,7 @@ package org.dbsyncer.biz.impl;
 import org.dbsyncer.biz.ConditionService;
 import org.dbsyncer.biz.vo.ConditionVo;
 import org.dbsyncer.connector.enums.FilterEnum;
-import org.dbsyncer.connector.enums.QuartzFilterEnum;
+import org.dbsyncer.sdk.enums.QuartzFilterEnum;
 import org.dbsyncer.parser.ProfileComponent;
 import org.dbsyncer.sdk.enums.OperationEnum;
 import org.springframework.stereotype.Component;

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

@@ -9,8 +9,8 @@ import org.dbsyncer.biz.model.MetricResponse;
 import org.dbsyncer.biz.model.MetricResponseInfo;
 import org.dbsyncer.biz.model.Sample;
 import org.dbsyncer.common.model.Paging;
-import org.dbsyncer.connector.scheduled.ScheduledTaskJob;
-import org.dbsyncer.connector.scheduled.ScheduledTaskService;
+import org.dbsyncer.sdk.scheduled.ScheduledTaskJob;
+import org.dbsyncer.sdk.scheduled.ScheduledTaskService;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.sdk.constant.ConnectorConstant;

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

@@ -26,8 +26,8 @@ import org.dbsyncer.common.util.JsonUtil;
 import org.dbsyncer.common.util.NumberUtil;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.enums.FilterEnum;
-import org.dbsyncer.connector.scheduled.ScheduledTaskJob;
-import org.dbsyncer.connector.scheduled.ScheduledTaskService;
+import org.dbsyncer.sdk.scheduled.ScheduledTaskJob;
+import org.dbsyncer.sdk.scheduled.ScheduledTaskService;
 import org.dbsyncer.parser.LogService;
 import org.dbsyncer.parser.LogType;
 import org.dbsyncer.parser.ProfileComponent;

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

@@ -1,7 +1,7 @@
 package org.dbsyncer.biz.vo;
 
 import org.dbsyncer.connector.enums.FilterEnum;
-import org.dbsyncer.connector.enums.QuartzFilterEnum;
+import org.dbsyncer.sdk.enums.QuartzFilterEnum;
 import org.dbsyncer.sdk.enums.OperationEnum;
 
 import java.util.List;

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

@@ -4,10 +4,10 @@ import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.JsonUtil;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.enums.QuartzFilterEnum;
-import org.dbsyncer.connector.quartz.AbstractQuartzListener;
-import org.dbsyncer.connector.quartz.Point;
-import org.dbsyncer.connector.quartz.QuartzFilter;
+import org.dbsyncer.sdk.enums.QuartzFilterEnum;
+import org.dbsyncer.sdk.listener.AbstractQuartzListener;
+import org.dbsyncer.sdk.quartz.Point;
+import org.dbsyncer.sdk.quartz.QuartzFilter;
 import org.dbsyncer.sdk.constant.ConnectorConstant;
 import org.dbsyncer.sdk.model.Filter;
 

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

@@ -1,15 +1,15 @@
 package org.dbsyncer.connector.file;
 
 import org.apache.commons.io.IOUtils;
-import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.sdk.model.ChangedOffset;
-import org.dbsyncer.sdk.listener.event.RowChangedEvent;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.NumberUtil;
 import org.dbsyncer.common.util.StringUtil;
+import org.dbsyncer.connector.ConnectorException;
 import org.dbsyncer.connector.config.FileConfig;
 import org.dbsyncer.sdk.constant.ConnectorConstant;
-import org.dbsyncer.connector.AbstractListener;
+import org.dbsyncer.sdk.listener.AbstractListener;
+import org.dbsyncer.sdk.listener.event.RowChangedEvent;
+import org.dbsyncer.sdk.model.ChangedOffset;
 import org.dbsyncer.sdk.model.Field;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -48,7 +48,7 @@ public class FileListener extends AbstractListener {
     private static final String CHARSET_NAME = "UTF-8";
     private final Lock connectLock = new ReentrantLock();
     private volatile boolean connected;
-    private FileConnectorInstance connectorInstance;
+    private FileConnectorInstance instance;
     private WatchService watchService;
     private Worker worker;
     private Map<String, PipelineResolver> pipeline = new ConcurrentHashMap<>();
@@ -64,9 +64,9 @@ public class FileListener extends AbstractListener {
                 return;
             }
 
-            connectorInstance = (FileConnectorInstance) connectorFactory.connect(connectorConfig);
-            final FileConfig config = connectorInstance.getConfig();
-            final String cacheKey = connectorFactory.getConnectorService(connectorConfig.getConnectorType()).getConnectorInstanceCacheKey(connectorConfig);
+            instance = (FileConnectorInstance) connectorInstance;
+            final FileConfig config = instance.getConfig();
+            final String cacheKey = connectorService.getConnectorInstanceCacheKey(config);
             connected = true;
 
             separator = config.getSeparator();
@@ -93,7 +93,7 @@ public class FileListener extends AbstractListener {
     }
 
     private void initPipeline(String fileDir) throws IOException {
-        for (FileSchema fileSchema : connectorInstance.getFileSchemaList()) {
+        for (FileSchema fileSchema : instance.getFileSchemaList()) {
             String fileName = fileSchema.getFileName();
             String file = fileDir.concat(fileName);
             Assert.isTrue(new File(file).exists(), String.format("found not file '%s'", file));

dbsyncer-connector/src/main/java/org/dbsyncer/connector/mysql/MySQLConnector.java

@@ -1,7 +1,7 @@
 package org.dbsyncer.connector.mysql;
 
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.quartz.DatabaseQuartzListener;
+import org.dbsyncer.sdk.quartz.DatabaseQuartzListener;
 import org.dbsyncer.sdk.config.ReaderConfig;
 import org.dbsyncer.sdk.connector.database.AbstractDatabaseConnector;
 import org.dbsyncer.sdk.constant.DatabaseConstant;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/mysql/MySQLListener.java

@@ -15,7 +15,7 @@ import net.sf.jsqlparser.JSQLParserException;
 import net.sf.jsqlparser.parser.CCJSqlParserUtil;
 import net.sf.jsqlparser.statement.alter.Alter;
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.AbstractDatabaseListener;
+import org.dbsyncer.sdk.listener.AbstractDatabaseListener;
 import org.dbsyncer.connector.ConnectorException;
 import org.dbsyncer.sdk.config.DatabaseConfig;
 import org.dbsyncer.sdk.constant.ConnectorConstant;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/oracle/OracleConnector.java

@@ -1,7 +1,7 @@
 package org.dbsyncer.connector.oracle;
 
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.quartz.DatabaseQuartzListener;
+import org.dbsyncer.sdk.quartz.DatabaseQuartzListener;
 import org.dbsyncer.sdk.config.CommandConfig;
 import org.dbsyncer.sdk.config.DatabaseConfig;
 import org.dbsyncer.sdk.config.ReaderConfig;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/oracle/OracleListener.java

@@ -1,6 +1,6 @@
 package org.dbsyncer.connector.oracle;
 
-import org.dbsyncer.connector.AbstractDatabaseListener;
+import org.dbsyncer.sdk.listener.AbstractDatabaseListener;
 import org.dbsyncer.connector.ConnectorException;
 import org.dbsyncer.connector.oracle.dcn.DBChangeNotification;
 import org.dbsyncer.sdk.config.DatabaseConfig;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/postgresql/MessageDecoder.java

@@ -1,9 +1,9 @@
 package org.dbsyncer.connector.postgresql;
 
-import org.dbsyncer.sdk.listener.event.RowChangedEvent;
-import org.dbsyncer.connector.ConnectorFactory;
 import org.dbsyncer.sdk.config.DatabaseConfig;
 import org.dbsyncer.sdk.connector.database.DatabaseConnectorInstance;
+import org.dbsyncer.sdk.listener.event.RowChangedEvent;
+import org.dbsyncer.sdk.spi.ConnectorService;
 import org.postgresql.replication.LogSequenceNumber;
 import org.postgresql.replication.fluent.logical.ChainedLogicalStreamBuilder;
 
@@ -16,7 +16,7 @@ import java.nio.ByteBuffer;
  */
 public interface MessageDecoder {
 
-    default void postProcessBeforeInitialization(ConnectorFactory connectorFactory, DatabaseConnectorInstance connectorInstance) {
+    default void postProcessBeforeInitialization(ConnectorService connectorService, DatabaseConnectorInstance connectorInstance) {
     }
 
     boolean skipMessage(ByteBuffer buffer, LogSequenceNumber startLsn, LogSequenceNumber lastReceiveLsn);

dbsyncer-connector/src/main/java/org/dbsyncer/connector/postgresql/PostgreSQLConnector.java

@@ -1,7 +1,7 @@
 package org.dbsyncer.connector.postgresql;
 
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.quartz.DatabaseQuartzListener;
+import org.dbsyncer.sdk.quartz.DatabaseQuartzListener;
 import org.dbsyncer.sdk.config.CommandConfig;
 import org.dbsyncer.sdk.config.DatabaseConfig;
 import org.dbsyncer.sdk.config.ReaderConfig;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/postgresql/PostgreSQLListener.java

@@ -1,7 +1,7 @@
 package org.dbsyncer.connector.postgresql;
 
 import org.dbsyncer.common.util.BooleanUtil;
-import org.dbsyncer.connector.AbstractDatabaseListener;
+import org.dbsyncer.sdk.listener.AbstractDatabaseListener;
 import org.dbsyncer.connector.ConnectorException;
 import org.dbsyncer.connector.postgresql.enums.MessageDecoderEnum;
 import org.dbsyncer.sdk.config.DatabaseConfig;
@@ -52,7 +52,7 @@ public class PostgreSQLListener extends AbstractDatabaseListener {
     private final Lock connectLock = new ReentrantLock();
     private volatile boolean connected;
     private DatabaseConfig config;
-    private DatabaseConnectorInstance connectorInstance;
+    private DatabaseConnectorInstance instance;
     private Connection connection;
     private PGReplicationStream stream;
     private boolean dropSlotOnClose;
@@ -70,15 +70,15 @@ public class PostgreSQLListener extends AbstractDatabaseListener {
                 return;
             }
 
-            connectorInstance = (DatabaseConnectorInstance) connectorFactory.connect(connectorConfig);
-            config = connectorInstance.getConfig();
+            instance = (DatabaseConnectorInstance) connectorInstance;
+            config = instance.getConfig();
 
-            final String walLevel = connectorInstance.execute(databaseTemplate -> databaseTemplate.queryForObject(GET_WAL_LEVEL, String.class));
+            final String walLevel = instance.execute(databaseTemplate -> databaseTemplate.queryForObject(GET_WAL_LEVEL, String.class));
             if (!DEFAULT_WAL_LEVEL.equals(walLevel)) {
                 throw new ConnectorException(String.format("Postgres server wal_level property must be \"%s\" but is: %s", DEFAULT_WAL_LEVEL, walLevel));
             }
 
-            final boolean hasAuth = connectorInstance.execute(databaseTemplate -> {
+            final boolean hasAuth = instance.execute(databaseTemplate -> {
                 Map rs = databaseTemplate.queryForMap(GET_ROLE);
                 Boolean login = (Boolean) rs.getOrDefault("login", false);
                 Boolean replication = (Boolean) rs.getOrDefault("replication", false);
@@ -91,11 +91,11 @@ public class PostgreSQLListener extends AbstractDatabaseListener {
                 throw new ConnectorException(String.format("Postgres roles LOGIN and REPLICATION are not assigned to user: %s", config.getUsername()));
             }
 
-            database = connectorInstance.execute(databaseTemplate -> databaseTemplate.queryForObject(GET_DATABASE, String.class));
+            database = instance.execute(databaseTemplate -> databaseTemplate.queryForObject(GET_DATABASE, String.class));
             messageDecoder = MessageDecoderEnum.getMessageDecoder(config.getProperty(PLUGIN_NAME));
             messageDecoder.setMetaId(metaId);
             messageDecoder.setConfig(config);
-            messageDecoder.postProcessBeforeInitialization(connectorFactory, connectorInstance);
+            messageDecoder.postProcessBeforeInitialization(connectorService, instance);
             dropSlotOnClose = BooleanUtil.toBoolean(config.getProperty(DROP_SLOT_ON_CLOSE, "true"));
 
             connect();
@@ -170,7 +170,7 @@ public class PostgreSQLListener extends AbstractDatabaseListener {
     private void createReplicationSlot(PGConnection pgConnection) throws SQLException {
         String slotName = messageDecoder.getSlotName();
         String plugin = messageDecoder.getOutputPlugin();
-        boolean existSlot = connectorInstance.execute(databaseTemplate -> databaseTemplate.queryForObject(GET_SLOT, new Object[]{database, slotName, plugin}, Integer.class) > 0);
+        boolean existSlot = instance.execute(databaseTemplate -> databaseTemplate.queryForObject(GET_SLOT, new Object[]{database, slotName, plugin}, Integer.class) > 0);
         if (!existSlot) {
             pgConnection.getReplicationAPI()
                     .createReplicationSlot()
@@ -184,7 +184,7 @@ public class PostgreSQLListener extends AbstractDatabaseListener {
         }
 
         if (!snapshot.containsKey(LSN_POSITION)) {
-            LogSequenceNumber lsn = connectorInstance.execute(databaseTemplate -> LogSequenceNumber.valueOf(databaseTemplate.queryForObject(GET_RESTART_LSN, new Object[]{database, slotName, plugin}, String.class)));
+            LogSequenceNumber lsn = instance.execute(databaseTemplate -> LogSequenceNumber.valueOf(databaseTemplate.queryForObject(GET_RESTART_LSN, new Object[]{database, slotName, plugin}, String.class)));
             if (null == lsn || lsn.asLong() == 0) {
                 throw new ConnectorException("No maximum LSN recorded in the database");
             }
@@ -204,7 +204,7 @@ public class PostgreSQLListener extends AbstractDatabaseListener {
         final int ATTEMPTS = 3;
         for (int i = 0; i < ATTEMPTS; i++) {
             try {
-                connectorInstance.execute(databaseTemplate -> {
+                instance.execute(databaseTemplate -> {
                     databaseTemplate.execute(String.format("select pg_drop_replication_slot('%s')", slotName));
                     return true;
                 });

dbsyncer-connector/src/main/java/org/dbsyncer/connector/postgresql/decoder/PgOutputMessageDecoder.java

@@ -2,7 +2,6 @@ package org.dbsyncer.connector.postgresql.decoder;
 
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.ConnectorFactory;
 import org.dbsyncer.connector.postgresql.AbstractMessageDecoder;
 import org.dbsyncer.connector.postgresql.enums.MessageDecoderEnum;
 import org.dbsyncer.connector.postgresql.enums.MessageTypeEnum;
@@ -10,6 +9,7 @@ import org.dbsyncer.sdk.connector.database.DatabaseConnectorInstance;
 import org.dbsyncer.sdk.listener.event.RowChangedEvent;
 import org.dbsyncer.sdk.model.Field;
 import org.dbsyncer.sdk.model.MetaInfo;
+import org.dbsyncer.sdk.spi.ConnectorService;
 import org.postgresql.replication.fluent.logical.ChainedLogicalStreamBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -34,12 +34,12 @@ public class PgOutputMessageDecoder extends AbstractMessageDecoder {
     private static final LocalDateTime PG_EPOCH = LocalDateTime.of(2000, 1, 1, 0, 0, 0);
     private static final String GET_TABLE_SCHEMA = "select t.oid,t.relname as tableName from pg_class t inner join (select ns.oid as nspoid, ns.nspname from pg_namespace ns where ns.nspname = '%s') as n on n.nspoid = t.relnamespace where relkind = 'r'";
     private static final Map<Integer, TableId> tables = new LinkedHashMap<>();
-    private ConnectorFactory connectorFactory;
+    private ConnectorService connectorService;
     private DatabaseConnectorInstance connectorInstance;
 
     @Override
-    public void postProcessBeforeInitialization(ConnectorFactory connectorFactory, DatabaseConnectorInstance connectorInstance) {
-        this.connectorFactory = connectorFactory;
+    public void postProcessBeforeInitialization(ConnectorService connectorService, DatabaseConnectorInstance connectorInstance) {
+        this.connectorService = connectorService;
         this.connectorInstance = connectorInstance;
         initPublication();
         readSchema();
@@ -123,7 +123,7 @@ public class PgOutputMessageDecoder extends AbstractMessageDecoder {
             schemas.forEach(map -> {
                 Long oid = (Long) map.get("oid");
                 String tableName = (String) map.get("tableName");
-                MetaInfo metaInfo = connectorFactory.getMetaInfo(connectorInstance, tableName);
+                MetaInfo metaInfo = connectorService.getMetaInfo(connectorInstance, tableName);
                 Assert.notEmpty(metaInfo.getColumn(), String.format("The table column for '%s' must not be empty.", tableName));
                 tables.put(oid.intValue(), new TableId(oid.intValue(), tableName, metaInfo.getColumn()));
             });
@@ -156,7 +156,7 @@ public class PgOutputMessageDecoder extends AbstractMessageDecoder {
             logger.warn("The column size of table '{}' is {}, but we has been received column size is {}.", tableId.tableName, tableId.fields.size(), nColumn);
 
             // The table schema has been changed, we should be get a new table schema from db.
-            MetaInfo metaInfo = connectorFactory.getMetaInfo(connectorInstance, tableId.tableName);
+            MetaInfo metaInfo = connectorService.getMetaInfo(connectorInstance, tableId.tableName);
             if (CollectionUtils.isEmpty(metaInfo.getColumn())) {
                 throw new ConnectorException(String.format("The table column for '%s' is empty.", tableId.tableName));
             }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sql/DQLMySQLConnector.java

@@ -1,7 +1,7 @@
 package org.dbsyncer.connector.sql;
 
 import org.dbsyncer.connector.mysql.DqlMySQLListener;
-import org.dbsyncer.connector.quartz.DatabaseQuartzListener;
+import org.dbsyncer.sdk.quartz.DatabaseQuartzListener;
 import org.dbsyncer.sdk.config.CommandConfig;
 import org.dbsyncer.sdk.config.ReaderConfig;
 import org.dbsyncer.sdk.connector.database.AbstractDQLConnector;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sql/DQLOracleConnector.java

@@ -1,7 +1,7 @@
 package org.dbsyncer.connector.sql;
 
 import org.dbsyncer.connector.oracle.DqlOracleListener;
-import org.dbsyncer.connector.quartz.DatabaseQuartzListener;
+import org.dbsyncer.sdk.quartz.DatabaseQuartzListener;
 import org.dbsyncer.sdk.config.ReaderConfig;
 import org.dbsyncer.sdk.connector.database.AbstractDQLConnector;
 import org.dbsyncer.sdk.constant.DatabaseConstant;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sql/DQLPostgreSQLConnector.java

@@ -1,7 +1,7 @@
 package org.dbsyncer.connector.sql;
 
 import org.dbsyncer.connector.postgresql.DqlPostgreSQLListener;
-import org.dbsyncer.connector.quartz.DatabaseQuartzListener;
+import org.dbsyncer.sdk.quartz.DatabaseQuartzListener;
 import org.dbsyncer.sdk.config.ReaderConfig;
 import org.dbsyncer.sdk.connector.database.AbstractDQLConnector;
 import org.dbsyncer.sdk.constant.DatabaseConstant;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sql/DQLSqlServerConnector.java

@@ -1,7 +1,7 @@
 package org.dbsyncer.connector.sql;
 
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.quartz.DatabaseQuartzListener;
+import org.dbsyncer.sdk.quartz.DatabaseQuartzListener;
 import org.dbsyncer.connector.sqlserver.DqlSqlServerListener;
 import org.dbsyncer.sdk.config.ReaderConfig;
 import org.dbsyncer.sdk.connector.database.AbstractDQLConnector;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sqlserver/SqlServerConnector.java

@@ -2,7 +2,7 @@ package org.dbsyncer.connector.sqlserver;
 
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.quartz.DatabaseQuartzListener;
+import org.dbsyncer.sdk.quartz.DatabaseQuartzListener;
 import org.dbsyncer.sdk.config.CommandConfig;
 import org.dbsyncer.sdk.config.DatabaseConfig;
 import org.dbsyncer.sdk.config.ReaderConfig;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sqlserver/SqlServerListener.java

@@ -3,13 +3,14 @@ package org.dbsyncer.connector.sqlserver;
 import com.microsoft.sqlserver.jdbc.SQLServerException;
 import org.dbsyncer.connector.ConnectorException;
 import org.dbsyncer.connector.enums.TableOperationEnum;
+import org.dbsyncer.sdk.connector.database.AbstractDatabaseConnector;
 import org.dbsyncer.sdk.model.ChangedOffset;
 import org.dbsyncer.sdk.listener.event.RowChangedEvent;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.sdk.config.DatabaseConfig;
 import org.dbsyncer.sdk.constant.ConnectorConstant;
 import org.dbsyncer.sdk.connector.database.DatabaseConnectorInstance;
-import org.dbsyncer.connector.AbstractDatabaseListener;
+import org.dbsyncer.sdk.listener.AbstractDatabaseListener;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.util.Assert;
@@ -62,7 +63,7 @@ public class SqlServerListener extends AbstractDatabaseListener {
     private volatile boolean connected;
     private Set<String> tables;
     private Set<SqlServerChangeTable> changeTables;
-    private DatabaseConnectorInstance connectorInstance;
+    private DatabaseConnectorInstance instance;
     private Worker worker;
     private Lsn lastLsn;
     private String serverName;
@@ -135,9 +136,10 @@ public class SqlServerListener extends AbstractDatabaseListener {
     }
 
     private void connect() {
-        if (connectorFactory.isAlive(connectorConfig)) {
-            connectorInstance = (DatabaseConnectorInstance) connectorFactory.connect(connectorConfig);
-            DatabaseConfig cfg = connectorInstance.getConfig();
+        instance = (DatabaseConnectorInstance) connectorInstance;
+        AbstractDatabaseConnector service = (AbstractDatabaseConnector) connectorService;
+        if (service.isAlive(instance)) {
+            DatabaseConfig cfg = instance.getConfig();
             serverName = cfg.getUrl();
             schema = cfg.getSchema();
         }
@@ -221,7 +223,7 @@ public class SqlServerListener extends AbstractDatabaseListener {
     }
 
     private void execute(String... sqlStatements) {
-        connectorInstance.execute(databaseTemplate -> {
+        instance.execute(databaseTemplate -> {
             for (String sqlStatement : sqlStatements) {
                 if (sqlStatement != null) {
                     logger.info("executing '{}'", sqlStatement);
@@ -313,7 +315,7 @@ public class SqlServerListener extends AbstractDatabaseListener {
     }
 
     private <T> T query(String preparedQuerySql, StatementPreparer statementPreparer, ResultSetMapper<T> mapper) {
-        Object execute = connectorInstance.execute(databaseTemplate -> {
+        Object execute = instance.execute(databaseTemplate -> {
             PreparedStatement ps = null;
             ResultSet rs = null;
             T apply = null;

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

@@ -1,12 +1,7 @@
 package org.dbsyncer.manager.impl;
 
 import org.dbsyncer.common.util.CollectionUtils;
-import org.dbsyncer.connector.AbstractListener;
 import org.dbsyncer.connector.ConnectorFactory;
-import org.dbsyncer.connector.quartz.AbstractQuartzListener;
-import org.dbsyncer.connector.quartz.TableGroupQuartzCommand;
-import org.dbsyncer.connector.scheduled.ScheduledTaskJob;
-import org.dbsyncer.connector.scheduled.ScheduledTaskService;
 import org.dbsyncer.manager.AbstractPuller;
 import org.dbsyncer.manager.ManagerException;
 import org.dbsyncer.parser.LogService;
@@ -22,10 +17,15 @@ import org.dbsyncer.parser.model.Meta;
 import org.dbsyncer.parser.model.TableGroup;
 import org.dbsyncer.sdk.config.ListenerConfig;
 import org.dbsyncer.sdk.enums.ListenerTypeEnum;
+import org.dbsyncer.sdk.listener.AbstractListener;
+import org.dbsyncer.sdk.listener.AbstractQuartzListener;
 import org.dbsyncer.sdk.listener.Listener;
 import org.dbsyncer.sdk.model.ChangedOffset;
 import org.dbsyncer.sdk.model.ConnectorConfig;
 import org.dbsyncer.sdk.model.Table;
+import org.dbsyncer.sdk.quartz.TableGroupQuartzCommand;
+import org.dbsyncer.sdk.scheduled.ScheduledTaskJob;
+import org.dbsyncer.sdk.scheduled.ScheduledTaskService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.context.ApplicationListener;
@@ -173,7 +173,8 @@ public final class IncrementPuller extends AbstractPuller implements Application
                 filterTable.add(table.getName());
             });
 
-            abstractListener.setConnectorFactory(connectorFactory);
+            abstractListener.setConnectorService(connectorFactory.getConnectorService(connectorConfig.getConnectorType()));
+            abstractListener.setConnectorInstance(connectorFactory.connect(connectorConfig));
             abstractListener.setScheduledTaskService(scheduledTaskService);
             abstractListener.setConnectorConfig(connectorConfig);
             abstractListener.setListenerConfig(listenerConfig);

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

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

dbsyncer-parser/src/main/java/org/dbsyncer/parser/flush/AbstractBufferActuator.java

@@ -1,8 +1,8 @@
 package org.dbsyncer.parser.flush;
 
 import org.dbsyncer.common.config.BufferActuatorConfig;
-import org.dbsyncer.connector.scheduled.ScheduledTaskJob;
-import org.dbsyncer.connector.scheduled.ScheduledTaskService;
+import org.dbsyncer.sdk.scheduled.ScheduledTaskJob;
+import org.dbsyncer.sdk.scheduled.ScheduledTaskService;
 import org.dbsyncer.parser.ProfileComponent;
 import org.dbsyncer.parser.enums.MetaEnum;
 import org.dbsyncer.parser.model.Meta;

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

@@ -1,7 +1,7 @@
 package org.dbsyncer.parser.flush.impl;
 
 import org.dbsyncer.common.config.TableGroupBufferConfig;
-import org.dbsyncer.connector.scheduled.ScheduledTaskService;
+import org.dbsyncer.sdk.scheduled.ScheduledTaskService;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.common.util.ThreadPoolUtil;
 import org.dbsyncer.common.util.UUIDUtil;

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

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

dbsyncer-connector/src/main/java/org/dbsyncer/connector/enums/QuartzFilterEnum.java → dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/enums/QuartzFilterEnum.java

@@ -1,11 +1,11 @@
-package org.dbsyncer.connector.enums;
+package org.dbsyncer.sdk.enums;
 
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.quartz.QuartzFilter;
-import org.dbsyncer.connector.quartz.filter.DateFilter;
-import org.dbsyncer.connector.quartz.filter.TimestampFilter;
-import org.dbsyncer.connector.quartz.filter.YesDateFilter;
-import org.dbsyncer.connector.quartz.filter.YesTimestampFilter;
+import org.dbsyncer.sdk.quartz.QuartzFilter;
+import org.dbsyncer.sdk.quartz.filter.DateFilter;
+import org.dbsyncer.sdk.quartz.filter.TimestampFilter;
+import org.dbsyncer.sdk.quartz.filter.YesDateFilter;
+import org.dbsyncer.sdk.quartz.filter.YesTimestampFilter;
 
 /**
  * @author AE86

dbsyncer-connector/src/main/java/org/dbsyncer/connector/AbstractDatabaseListener.java → dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/listener/AbstractDatabaseListener.java

@@ -1,11 +1,10 @@
-package org.dbsyncer.connector;
+package org.dbsyncer.sdk.listener;
 
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.sdk.connector.database.AbstractDatabaseConnector;
 import org.dbsyncer.sdk.connector.database.DatabaseConnectorInstance;
 import org.dbsyncer.sdk.constant.ConnectorConstant;
-import org.dbsyncer.sdk.listener.ChangedEvent;
 import org.dbsyncer.sdk.listener.event.RowChangedEvent;
 import org.dbsyncer.sdk.model.Field;
 import org.dbsyncer.sdk.model.MetaInfo;
@@ -78,13 +77,13 @@ public abstract class AbstractDatabaseListener extends AbstractListener {
      * 初始化Dql连接配置
      */
     protected void postProcessDqlBeforeInitialization() {
-        DatabaseConnectorInstance connectorInstance = (DatabaseConnectorInstance) connectorFactory.connect(connectorConfig);
-        AbstractDatabaseConnector connectorService = (AbstractDatabaseConnector) connectorFactory.getConnectorService(connectorConfig.getConnectorType());
-        String quotation = connectorService.buildSqlWithQuotation();
+        DatabaseConnectorInstance instance = (DatabaseConnectorInstance) connectorInstance;
+        AbstractDatabaseConnector service = (AbstractDatabaseConnector) connectorService;
+        String quotation = service.buildSqlWithQuotation();
 
         // <用户表, MY_USER>
         Map<String, String> tableMap = new HashMap<>();
-        connectorInstance.getConfig().getSqlTables().forEach(s -> tableMap.put(s.getSqlName(), s.getTable()));
+        instance.getConfig().getSqlTables().forEach(s -> tableMap.put(s.getSqlName(), s.getTable()));
         // 清空默认表名
         filterTable.clear();
         for (Table t : sourceTable) {
@@ -95,7 +94,7 @@ public abstract class AbstractDatabaseListener extends AbstractListener {
             Assert.hasText(sql, "The sql is null.");
             Assert.hasText(tableName, "The tableName is null.");
 
-            MetaInfo metaInfo = connectorFactory.getMetaInfo(connectorInstance, sqlName);
+            MetaInfo metaInfo = service.getMetaInfo(instance, sqlName);
             final List<Field> column = metaInfo.getColumn();
             Assert.notEmpty(column, String.format("The column of table name '%s' is empty.", sqlName));
 
@@ -107,7 +106,7 @@ public abstract class AbstractDatabaseListener extends AbstractListener {
             boolean notContainsWhere = !StringUtil.contains(sql, " WHERE ");
             querySql.append(notContainsWhere ? " WHERE " : " AND ");
             PrimaryKeyUtil.buildSql(querySql, primaryKeys, quotation, " AND ", " = ? ", notContainsWhere);
-            DqlMapper dqlMapper = new DqlMapper(connectorInstance, sqlName, querySql.toString(), column, getPrimaryKeyIndexArray(column, primaryKeys));
+            DqlMapper dqlMapper = new DqlMapper(instance, sqlName, querySql.toString(), column, getPrimaryKeyIndexArray(column, primaryKeys));
             if (!dqlMap.containsKey(tableName)) {
                 dqlMap.putIfAbsent(tableName, new ArrayList<>());
             }
@@ -140,7 +139,7 @@ public abstract class AbstractDatabaseListener extends AbstractListener {
 
     private void queryDqlData(DqlMapper dqlMapper, List<Object> data) {
         if (!CollectionUtils.isEmpty(data)) {
-            Map<String, Object> row = dqlMapper.connectorInstance.execute(databaseTemplate -> {
+            Map<String, Object> row = dqlMapper.instance.execute(databaseTemplate -> {
                 int size = dqlMapper.primaryKeyIndexArray.length;
                 Object[] args = new Object[size];
                 for (int i = 0; i < size; i++) {
@@ -156,14 +155,14 @@ public abstract class AbstractDatabaseListener extends AbstractListener {
     }
 
     final class DqlMapper {
-        DatabaseConnectorInstance connectorInstance;
+        DatabaseConnectorInstance instance;
         String sqlName;
         String sql;
         List<Field> column;
         Integer[] primaryKeyIndexArray;
 
-        public DqlMapper(DatabaseConnectorInstance connectorInstance, String sqlName, String sql, List<Field> column, Integer[] primaryKeyIndexArray) {
-            this.connectorInstance = connectorInstance;
+        public DqlMapper(DatabaseConnectorInstance instance, String sqlName, String sql, List<Field> column, Integer[] primaryKeyIndexArray) {
+            this.instance = instance;
             this.sqlName = sqlName;
             this.sql = sql;
             this.column = column;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/AbstractListener.java → dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/listener/AbstractListener.java

@@ -1,15 +1,14 @@
-package org.dbsyncer.connector;
+package org.dbsyncer.sdk.listener;
 
 import org.dbsyncer.common.util.CollectionUtils;
-import org.dbsyncer.connector.scheduled.ScheduledTaskService;
+import org.dbsyncer.sdk.scheduled.ScheduledTaskService;
 import org.dbsyncer.sdk.config.ListenerConfig;
+import org.dbsyncer.sdk.connector.ConnectorInstance;
 import org.dbsyncer.sdk.constant.ConnectorConstant;
-import org.dbsyncer.sdk.listener.ChangedEvent;
-import org.dbsyncer.sdk.listener.Listener;
-import org.dbsyncer.sdk.listener.Watcher;
 import org.dbsyncer.sdk.model.ChangedOffset;
 import org.dbsyncer.sdk.model.ConnectorConfig;
 import org.dbsyncer.sdk.model.Table;
+import org.dbsyncer.sdk.spi.ConnectorService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -29,7 +28,8 @@ public abstract class AbstractListener implements Listener {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
     private final int FLUSH_DELAYED_SECONDS = 20;
-    protected ConnectorFactory connectorFactory;
+    protected ConnectorInstance connectorInstance;
+    protected ConnectorService connectorService;
     protected ScheduledTaskService scheduledTaskService;
     protected ConnectorConfig connectorConfig;
     protected ListenerConfig listenerConfig;
@@ -119,8 +119,12 @@ public abstract class AbstractListener implements Listener {
         }
     }
 
-    public void setConnectorFactory(ConnectorFactory connectorFactory) {
-        this.connectorFactory = connectorFactory;
+    public void setConnectorInstance(ConnectorInstance connectorInstance) {
+        this.connectorInstance = connectorInstance;
+    }
+
+    public void setConnectorService(ConnectorService connectorService) {
+        this.connectorService = connectorService;
     }
 
     public void setScheduledTaskService(ScheduledTaskService scheduledTaskService) {

dbsyncer-connector/src/main/java/org/dbsyncer/connector/quartz/AbstractQuartzListener.java → dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/listener/AbstractQuartzListener.java

@@ -1,16 +1,16 @@
-package org.dbsyncer.connector.quartz;
+package org.dbsyncer.sdk.listener;
 
 import org.dbsyncer.common.model.Result;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.common.util.UUIDUtil;
-import org.dbsyncer.connector.AbstractListener;
-import org.dbsyncer.connector.scheduled.ScheduledTaskJob;
 import org.dbsyncer.sdk.config.ReaderConfig;
-import org.dbsyncer.sdk.connector.ConnectorInstance;
 import org.dbsyncer.sdk.constant.ConnectorConstant;
 import org.dbsyncer.sdk.listener.event.ScanChangedEvent;
 import org.dbsyncer.sdk.model.Table;
+import org.dbsyncer.sdk.quartz.Point;
+import org.dbsyncer.sdk.quartz.TableGroupQuartzCommand;
+import org.dbsyncer.sdk.scheduled.ScheduledTaskJob;
 import org.dbsyncer.sdk.util.PrimaryKeyUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -103,14 +103,13 @@ public abstract class AbstractQuartzListener extends AbstractListener implements
         boolean supportedCursor = StringUtil.isNotBlank(command.get(ConnectorConstant.OPERTION_QUERY_CURSOR));
 
         // 检查增量点
-        ConnectorInstance connectionInstance = connectorFactory.connect(connectorConfig);
         Point point = checkLastPoint(command, index);
         int pageIndex = 1;
         Object[] cursors = PrimaryKeyUtil.getLastCursors(snapshot.get(index + CURSOR));
 
         while (running) {
             ReaderConfig readerConfig = new ReaderConfig(table, point.getCommand(), point.getArgs(), supportedCursor, cursors, pageIndex++, READ_NUM);
-            Result reader = connectorFactory.reader(connectionInstance, readerConfig);
+            Result reader = connectorService.reader(connectorInstance, readerConfig);
             List<Map> data = reader.getSuccessData();
             if (CollectionUtils.isEmpty(data)) {
                 cursors = new Object[0];

dbsyncer-connector/src/main/java/org/dbsyncer/connector/quartz/DatabaseQuartzListener.java → dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/quartz/DatabaseQuartzListener.java

@@ -1,9 +1,10 @@
-package org.dbsyncer.connector.quartz;
+package org.dbsyncer.sdk.quartz;
 
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.enums.QuartzFilterEnum;
+import org.dbsyncer.sdk.enums.QuartzFilterEnum;
 import org.dbsyncer.sdk.constant.ConnectorConstant;
+import org.dbsyncer.sdk.listener.AbstractQuartzListener;
 import org.springframework.util.Assert;
 
 import java.util.ArrayList;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/quartz/Point.java → dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/quartz/Point.java

@@ -1,4 +1,4 @@
-package org.dbsyncer.connector.quartz;
+package org.dbsyncer.sdk.quartz;
 
 import org.dbsyncer.common.util.StringUtil;
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/quartz/QuartzFilter.java → dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/quartz/QuartzFilter.java

@@ -1,4 +1,4 @@
-package org.dbsyncer.connector.quartz;
+package org.dbsyncer.sdk.quartz;
 
 public interface QuartzFilter<T> {
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/quartz/TableGroupQuartzCommand.java → dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/quartz/TableGroupQuartzCommand.java

@@ -1,4 +1,4 @@
-package org.dbsyncer.connector.quartz;
+package org.dbsyncer.sdk.quartz;
 
 import org.dbsyncer.sdk.model.Table;
 import org.dbsyncer.sdk.util.PrimaryKeyUtil;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/quartz/filter/DateFilter.java → dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/quartz/filter/DateFilter.java

@@ -1,7 +1,7 @@
-package org.dbsyncer.connector.quartz.filter;
+package org.dbsyncer.sdk.quartz.filter;
 
 import org.dbsyncer.common.util.DateFormatUtil;
-import org.dbsyncer.connector.quartz.QuartzFilter;
+import org.dbsyncer.sdk.quartz.QuartzFilter;
 
 import java.sql.Date;
 import java.time.Instant;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/quartz/filter/TimestampFilter.java → dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/quartz/filter/TimestampFilter.java

@@ -1,6 +1,6 @@
-package org.dbsyncer.connector.quartz.filter;
+package org.dbsyncer.sdk.quartz.filter;
 
-import org.dbsyncer.connector.quartz.QuartzFilter;
+import org.dbsyncer.sdk.quartz.QuartzFilter;
 
 import java.sql.Timestamp;
 import java.time.Instant;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/quartz/filter/YesDateFilter.java → dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/quartz/filter/YesDateFilter.java

@@ -1,7 +1,7 @@
-package org.dbsyncer.connector.quartz.filter;
+package org.dbsyncer.sdk.quartz.filter;
 
 import org.dbsyncer.common.util.DateFormatUtil;
-import org.dbsyncer.connector.quartz.QuartzFilter;
+import org.dbsyncer.sdk.quartz.QuartzFilter;
 
 import java.sql.Date;
 import java.time.LocalDateTime;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/quartz/filter/YesTimestampFilter.java → dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/quartz/filter/YesTimestampFilter.java

@@ -1,6 +1,6 @@
-package org.dbsyncer.connector.quartz.filter;
+package org.dbsyncer.sdk.quartz.filter;
 
-import org.dbsyncer.connector.quartz.QuartzFilter;
+import org.dbsyncer.sdk.quartz.QuartzFilter;
 
 import java.sql.Timestamp;
 import java.time.LocalDateTime;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/scheduled/ScheduledTaskJob.java → dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/scheduled/ScheduledTaskJob.java

@@ -1,4 +1,4 @@
-package org.dbsyncer.connector.scheduled;
+package org.dbsyncer.sdk.scheduled;
 
 /**
  * @version 1.0.0

dbsyncer-connector/src/main/java/org/dbsyncer/connector/scheduled/ScheduledTaskService.java → dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/scheduled/ScheduledTaskService.java

@@ -1,4 +1,4 @@
-package org.dbsyncer.connector.scheduled;
+package org.dbsyncer.sdk.scheduled;
 
 public interface ScheduledTaskService {
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/scheduled/impl/ScheduledTaskServiceImpl.java → dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/scheduled/impl/ScheduledTaskServiceImpl.java

@@ -1,8 +1,8 @@
-package org.dbsyncer.connector.scheduled.impl;
+package org.dbsyncer.sdk.scheduled.impl;
 
 import org.dbsyncer.common.CommonException;
-import org.dbsyncer.connector.scheduled.ScheduledTaskJob;
-import org.dbsyncer.connector.scheduled.ScheduledTaskService;
+import org.dbsyncer.sdk.scheduled.ScheduledTaskJob;
+import org.dbsyncer.sdk.scheduled.ScheduledTaskService;
 import org.dbsyncer.common.util.UUIDUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

dbsyncer-web/src/main/resources/application.properties

@@ -46,8 +46,8 @@ dbsyncer.parser.table.group.buffer-period-millisecond=300
 #storage
 # 是否使用MySQL存储配置（false-关闭; true-开启）
 # false: 保存磁盘/data/config(驱动配置)|data(按驱动分别存储增量数据)|log(系统日志)}
-dbsyncer.storage.support.mysql.enabled=false
-dbsyncer.storage.support.mysql.config.url=jdbc:mysql://127.0.0.1:3306/dbsyncer?rewriteBatchedStatements=true&seUnicode=true&characterEncoding=UTF8&serverTimezone=Asia/Shanghai&useSSL=false&verifyServerCertificate=false&autoReconnect=true
+dbsyncer.storage.support.mysql.enabled=true
+dbsyncer.storage.support.mysql.config.url=jdbc:mysql://127.0.0.1:3305/dbsyncer?rewriteBatchedStatements=true&seUnicode=true&characterEncoding=UTF8&serverTimezone=Asia/Shanghai&useSSL=false&verifyServerCertificate=false&autoReconnect=true
 dbsyncer.storage.support.mysql.config.username=root
 dbsyncer.storage.support.mysql.config.password=123
 # [StorageBufferActuator]线程数