impl es -> es; es -> db; db -> es

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

@@ -84,7 +84,7 @@ public final class ESConnector extends AbstractConnector implements Connector<ES
 
     @Override
     public String getConnectorMapperCacheKey(ESConfig config) {
-        return String.format("%s-%s", config.getUrl(), config.getUsername());
+        return String.format("%s-%s-%s-%s", config.getUrl(), config.getIndex(), config.getType(), config.getUsername());
     }
 
     @Override
@@ -272,11 +272,12 @@ public final class ESConnector extends AbstractConnector implements Connector<ES
 
         // 过滤条件
         String filterJson = command.get(ConnectorConstant.OPERTION_QUERY_FILTER);
-        if (StringUtil.isBlank(filterJson)) {
-            return;
+        List<Filter> filters = null;
+        if (!StringUtil.isBlank(filterJson)) {
+            filters = JsonUtil.jsonToArray(filterJson, Filter.class);
         }
-        List<Filter> filters = JsonUtil.jsonToArray(filterJson, Filter.class);
         if (CollectionUtils.isEmpty(filters)) {
+            builder.query(QueryBuilders.matchAllQuery());
             return;
         }
 

dbsyncer-listener/src/main/test/ESClientTest.java

@@ -36,7 +36,6 @@ import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.util.Arrays;
-import java.util.Date;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -169,7 +168,7 @@ public class ESClientTest {
         map.put("name", "刘备关羽张飞");
         map.put("content", "桃园结义");
         map.put("tags", new Long[]{200L});
-        map.put("update_time", new Date());
+        map.put("update_time", System.currentTimeMillis());
         IndexRequest request = new IndexRequest(indexName, type, "2");
         request.source(map, XContentType.JSON);
 
@@ -186,7 +185,7 @@ public class ESClientTest {
         m1.put("name", "刘备关羽张飞");
         m1.put("content", "桃园结义");
         m1.put("tags", new Long[]{200L});
-        m1.put("update_time", new Date());
+        m1.put("update_time", System.currentTimeMillis());
         IndexRequest r1 = new IndexRequest(indexName, type, "1");
         r1.source(m1, XContentType.JSON);
         request.add(r1);
@@ -196,7 +195,7 @@ public class ESClientTest {
         m2.put("name", "曹阿瞒");
         m2.put("content", "火烧");
         m2.put("tags", new Long[]{200L, 300L});
-        m2.put("update_time", new Date());
+        m2.put("update_time", System.currentTimeMillis());
         IndexRequest r2 = new IndexRequest(indexName, type, "2");
         r2.source(m2, XContentType.JSON);
         request.add(r2);

dbsyncer-parser/src/main/java/org/dbsyncer/parser/convert/handler/DateHandler.java

@@ -6,6 +6,8 @@ import org.dbsyncer.parser.convert.Handler;
 import java.util.Date;
 
 /**
+ * 系统日期
+ *
  * @author AE86
  * @version 1.0.0
  * @date 2019/10/8 23:03

dbsyncer-parser/src/main/java/org/dbsyncer/parser/convert/handler/LongToTimestampHandler.java

@@ -0,0 +1,25 @@
+package org.dbsyncer.parser.convert.handler;
+
+import org.dbsyncer.parser.convert.AbstractHandler;
+
+import java.sql.Timestamp;
+
+/**
+ * Long转Timestamp
+ *
+ * @author AE86
+ * @version 1.0.0
+ * @date 2021/9/2 23:04
+ */
+public class LongToTimestampHandler extends AbstractHandler {
+
+    @Override
+    public Object convert(String args, Object value) {
+        if (value instanceof Long) {
+            Long l = (Long) value;
+            value = new Timestamp(l);
+        }
+        return value;
+    }
+
+}

dbsyncer-parser/src/main/java/org/dbsyncer/parser/convert/handler/DefaultTimestampHandler.java → dbsyncer-parser/src/main/java/org/dbsyncer/parser/convert/handler/TimestampHandler.java

@@ -7,11 +7,13 @@ import java.sql.Timestamp;
 import java.time.Instant;
 
 /**
+ * 系统时间戳
+ *
  * @author AE86
  * @version 1.0.0
  * @date 2019/10/8 23:03
  */
-public class DefaultTimestampHandler implements Handler {
+public class TimestampHandler implements Handler {
 
     @Override
     public Object handle(String args, Object value) {

dbsyncer-parser/src/main/java/org/dbsyncer/parser/convert/handler/TimestampToLongHandler.java

@@ -0,0 +1,25 @@
+package org.dbsyncer.parser.convert.handler;
+
+import org.dbsyncer.parser.convert.AbstractHandler;
+
+import java.sql.Timestamp;
+
+/**
+ * Timestamp转Long
+ *
+ * @author AE86
+ * @version 1.0.0
+ * @date 2021/9/2 23:04
+ */
+public class TimestampToLongHandler extends AbstractHandler {
+
+    @Override
+    public Object convert(String args, Object value) {
+        if (value instanceof Timestamp) {
+            Timestamp t = (Timestamp) value;
+            value = t.getTime();
+        }
+        return value;
+    }
+
+}

dbsyncer-parser/src/main/java/org/dbsyncer/parser/enums/ConvertEnum.java

@@ -19,17 +19,25 @@ public enum ConvertEnum {
      */
     DEFAULT("DEFAULT", "默认值", 1, new DefaultHandler()),
     /**
-     * 默认系统时间戳
+     * 系统时间戳
      */
-    DEFAULT_SYSTEM_TIMESTAMP("DEFAULT_SYSTEM_TIMESTAMP", "默认系统时间戳", 0, new DefaultTimestampHandler()),
+    SYSTEM_TIMESTAMP("SYSTEM_TIMESTAMP", "系统时间戳", 0, new TimestampHandler()),
     /**
-     * 默认系统日期
+     * 系统日期date
      */
-    SYSTEM_DATE("SYSTEM_DATE", "默认系统日期", 0, new DateHandler()),
+    SYSTEM_DATE("SYSTEM_DATE", "系统日期", 0, new DateHandler()),
     /**
-     * 时间戳转日期
+     * Timestamp转date
      */
-    TIMESTAMP_TO_DATE("TIMESTAMP_TO_DATE", "时间戳转日期", 0, new TimestampToDateHandler()),
+    TIMESTAMP_TO_DATE("TIMESTAMP_TO_DATE", "Timestamp转date", 0, new TimestampToDateHandler()),
+    /**
+     * Timestamp转Long
+     */
+    TIMESTAMP_TO_LONG("TIMESTAMP_TO_LONG", "Timestamp转Long", 0, new TimestampToLongHandler()),
+    /**
+     * Long转Timestamp
+     */
+    LONG_TO_TIMESTAMP("LONG_TO_TIMESTAMP", "Long转Timestamp", 0, new LongToTimestampHandler()),
     /**
      * 替换
      */