支持定时采集dql连接器

Signed-off-by: AE86 <836391306@qq.com>

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

@@ -8,11 +8,12 @@ import org.springframework.stereotype.Component;
 import javax.annotation.PostConstruct;
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.function.Function;
 
 @Component
 public class ListenerFactory implements Listener {
 
-    private Map<ListenerTypeEnum, ExtractorMapper> map = new LinkedHashMap<>();
+    private Map<ListenerTypeEnum, Function<String, Class>> map = new LinkedHashMap<>();
 
     @PostConstruct
     private void init() {
@@ -22,17 +23,13 @@ public class ListenerFactory implements Listener {
 
     @Override
     public <T> T getExtractor(ListenerTypeEnum listenerTypeEnum, String connectorType, Class<T> valueType) throws IllegalAccessException, InstantiationException {
-        ExtractorMapper mapper = map.get(listenerTypeEnum);
-        if (null == mapper) {
+        Function function = map.get(listenerTypeEnum);
+        if (null == function) {
             throw new ListenerException(String.format("Unsupported type \"%s\" for extractor \"%s\".", listenerTypeEnum, connectorType));
         }
 
-        Class<T> clazz = (Class<T>) mapper.getExtractor(connectorType);
+        Class<T> clazz = (Class<T>) function.apply(connectorType);
         return clazz.newInstance();
     }
 
-    interface ExtractorMapper {
-        Class getExtractor(String connectorType);
-    }
-
 }

dbsyncer-listener/src/main/java/org/dbsyncer/listener/enums/TimingExtractorEnum.java

@@ -34,7 +34,23 @@ public enum TimingExtractorEnum {
     /**
      * Elasticsearch
      */
-    ELASTIC_SEARCH(ConnectorEnum.ELASTIC_SEARCH.getType(), ESQuartzExtractor.class);
+    ELASTIC_SEARCH(ConnectorEnum.ELASTIC_SEARCH.getType(), ESQuartzExtractor.class),
+    /**
+     * DqlMysql
+     */
+    DQL_MYSQL(ConnectorEnum.DQL_MYSQL.getType(), DatabaseQuartzExtractor.class),
+    /**
+     * DqlOracle
+     */
+    DQL_ORACLE(ConnectorEnum.DQL_ORACLE.getType(), DatabaseQuartzExtractor.class),
+    /**
+     * DqlSqlServer
+     */
+    DQL_SQL_SERVER(ConnectorEnum.DQL_SQL_SERVER.getType(), DatabaseQuartzExtractor.class),
+    /**
+     * DqlPostgreSQL
+     */
+    DQL_POSTGRE_SQL(ConnectorEnum.DQL_POSTGRE_SQL.getType(), DatabaseQuartzExtractor.class);
 
     private String type;
     private Class clazz;