update

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

@@ -1,9 +1,4 @@
 package org.dbsyncer.listener;
 
-import org.dbsyncer.common.model.Task;
-import org.dbsyncer.connector.config.ConnectorConfig;
-import org.dbsyncer.listener.config.ListenerConfig;
-
 public interface Listener {
-    void execute(Task t, ListenerConfig listenerConfig, ConnectorConfig config);
 }

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

@@ -1,15 +1,8 @@
 package org.dbsyncer.listener;
 
 
-import org.dbsyncer.common.model.Task;
-import org.dbsyncer.connector.config.ConnectorConfig;
-import org.dbsyncer.listener.config.ListenerConfig;
 import org.springframework.stereotype.Component;
 
 @Component
 public class ListenerFactory implements Listener{
-
-    public void execute(Task task, ListenerConfig listenerConfig, ConnectorConfig connectorConfig) {
-        // extract
-    }
 }

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

@@ -250,7 +250,7 @@ public class ManagerFactory implements Manager, ApplicationContextAware, Applica
     public void changeMetaState(String metaId, MetaEnum metaEnum) {
         Meta meta = getMeta(metaId);
         int code = metaEnum.getCode();
-        if (meta.getState() != code) {
+        if (null != meta && meta.getState() != code) {
             meta.setState(code);
             meta.setUpdateTime(System.currentTimeMillis());
             editMeta(meta);

dbsyncer-manager/src/main/java/org/dbsyncer/manager/enums/TaskEnum.java

@@ -3,8 +3,8 @@ package org.dbsyncer.manager.enums;
 import org.apache.commons.lang.StringUtils;
 import org.dbsyncer.common.model.Task;
 import org.dbsyncer.manager.ManagerException;
-import org.dbsyncer.manager.extractor.task.LogTask;
-import org.dbsyncer.manager.extractor.task.TimingTask;
+import org.dbsyncer.manager.extractor.increment.LogIncrement;
+import org.dbsyncer.manager.extractor.increment.TimingIncrement;
 
 /**
  * @author AE86
@@ -16,11 +16,11 @@ public enum TaskEnum {
     /**
      * 定时
      */
-    TIMING("timing", new TimingTask()),
+    TIMING("timing", new TimingIncrement()),
     /**
      * 监听日志
      */
-    LOG("log", new LogTask());
+    LOG("log", new LogIncrement());
 
     private String type;
     private Task task;

dbsyncer-manager/src/main/java/org/dbsyncer/manager/extractor/Increment.java

@@ -0,0 +1,4 @@
+package org.dbsyncer.manager.extractor;
+
+public interface Increment {
+}

dbsyncer-manager/src/main/java/org/dbsyncer/manager/extractor/impl/IncrementExtractor.java

@@ -2,17 +2,22 @@ package org.dbsyncer.manager.extractor.impl;
 
 import org.dbsyncer.common.event.IncrementRefreshEvent;
 import org.dbsyncer.common.model.Task;
+import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.listener.Listener;
 import org.dbsyncer.listener.config.ListenerConfig;
 import org.dbsyncer.manager.Manager;
 import org.dbsyncer.manager.enums.TaskEnum;
 import org.dbsyncer.manager.extractor.AbstractExtractor;
+import org.dbsyncer.manager.extractor.Increment;
 import org.dbsyncer.parser.model.Connector;
 import org.dbsyncer.parser.model.Mapping;
 import org.dbsyncer.parser.model.Meta;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.ApplicationListener;
 import org.springframework.stereotype.Component;
 import org.springframework.util.Assert;
@@ -28,7 +33,7 @@ import java.util.concurrent.ConcurrentHashMap;
  * @date 2020/04/26 15:28
  */
 @Component
-public class IncrementExtractor extends AbstractExtractor implements ApplicationListener<IncrementRefreshEvent> {
+public class IncrementExtractor extends AbstractExtractor implements ApplicationContextAware, ApplicationListener<IncrementRefreshEvent> {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
@@ -40,26 +45,30 @@ public class IncrementExtractor extends AbstractExtractor implements Application
 
     private Map<String, Task> map = new ConcurrentHashMap<>();
 
+    private Map<String, Increment> handle;
+
+    @Override
+    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+        handle = applicationContext.getBeansOfType(Increment.class);
+    }
+
     @Override
     public void asyncStart(Mapping mapping) {
         ListenerConfig listenerConfig = mapping.getListener();
         Connector connector = manager.getConnector(mapping.getSourceConnectorId());
         Assert.notNull(connector, "连接器不能为空.");
         // log/timing
-        String type = listenerConfig.getListenerType();
-        Task task = TaskEnum.getIncrementTask(type);
-        Assert.notNull(task, "未知的增量同步方式.");
+        String type = StringUtil.toLowerCaseFirstOne(listenerConfig.getListenerType()).concat("Increment");
+        Increment increment = handle.get(type);
+        Assert.notNull(increment, "未知的增量同步方式.");
 
         final String metaId = mapping.getMetaId();
-        task.setId(metaId);
-        map.putIfAbsent(metaId, task);
+        map.putIfAbsent(metaId, new Task(metaId));
 
         try {
             // 执行任务
             logger.info("启动任务:{}", metaId);
-            Task t = map.get(metaId);
-            listener.execute(t, listenerConfig, connector.getConfig());
-
+            // TODO increment
         } catch (Exception e) {
             // TODO 记录错误日志
             logger.error(e.getMessage());

dbsyncer-manager/src/main/java/org/dbsyncer/manager/extractor/increment/LogIncrement.java

@@ -0,0 +1,11 @@
+package org.dbsyncer.manager.extractor.increment;
+
+import org.dbsyncer.manager.extractor.Increment;
+
+/**
+ * @version 1.0.0
+ * @Author AE86
+ * @Date 2020-05-08 00:31
+ */
+public class LogIncrement implements Increment {
+}

dbsyncer-manager/src/main/java/org/dbsyncer/manager/extractor/increment/TimingIncrement.java

@@ -0,0 +1,11 @@
+package org.dbsyncer.manager.extractor.increment;
+
+import org.dbsyncer.manager.extractor.Increment;
+
+/**
+ * @version 1.0.0
+ * @Author AE86
+ * @Date 2020-05-08 00:32
+ */
+public class TimingIncrement implements Increment {
+}

dbsyncer-manager/src/main/java/org/dbsyncer/manager/extractor/task/LogTask.java

@@ -1,11 +0,0 @@
-package org.dbsyncer.manager.extractor.task;
-
-import org.dbsyncer.common.model.Task;
-
-/**
- * @version 1.0.0
- * @Author AE86
- * @Date 2020-05-08 00:31
- */
-public class LogTask extends Task {
-}

dbsyncer-manager/src/main/java/org/dbsyncer/manager/extractor/task/TimingTask.java

@@ -1,11 +0,0 @@
-package org.dbsyncer.manager.extractor.task;
-
-import org.dbsyncer.common.model.Task;
-
-/**
- * @version 1.0.0
- * @Author AE86
- * @Date 2020-05-08 00:32
- */
-public class TimingTask extends Task {
-}