add convert handler

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

@@ -34,7 +34,6 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.TimeUnit;
 
 /**
  * @author AE86
@@ -184,7 +183,7 @@ public class ParserFactory implements Parser {
 
         for (; ; ) {
             if (!task.isRunning()) {
-                logger.warn("任务被终止:{}", metaId);
+                logger.warn("任务被中止:{}", metaId);
                 break;
             }
 
@@ -226,7 +225,7 @@ public class ParserFactory implements Parser {
      * @param writer
      * @param total
      */
-    private void flush(Task task, Result writer, int total) {
+    private void flush(Task task, Result writer, long total) {
         // 引用传递
         long fail = writer.getFail().get();
         long success = total - fail;

dbsyncer-parser/src/main/java/org/dbsyncer/parser/convert/AbstractHandler.java

@@ -0,0 +1,27 @@
+package org.dbsyncer.parser.convert;
+
+import org.dbsyncer.parser.ParserException;
+
+public abstract class AbstractHandler implements Handler {
+
+    /**
+     * 交给实现handler
+     *
+     * @param args  参数
+     * @param value 值
+     * @return
+     */
+    protected abstract Object convert(String args, Object value) throws Exception;
+
+    @Override
+    public Object handle(String args, Object value) {
+        if (null != value) {
+            try {
+                return convert(args, value);
+            } catch (Exception e) {
+                throw new ParserException(e.getMessage());
+            }
+        }
+        return null;
+    }
+}

dbsyncer-parser/src/main/java/org/dbsyncer/parser/convert/Handler.java

@@ -6,4 +6,13 @@ package org.dbsyncer.parser.convert;
  * @date 2019/10/8 22:55
  */
 public interface Handler {
-}
+
+    /**
+     * 值转换
+     *
+     * @param args 参数
+     * @param value 值
+     * @return
+     */
+    Object handle(String args, Object value);
+}

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

@@ -1,11 +1,20 @@
 package org.dbsyncer.parser.convert.handler;
 
-import org.dbsyncer.parser.convert.Handler;
+import org.dbsyncer.common.util.AESEncyptUtil;
+import org.dbsyncer.parser.convert.AbstractHandler;
 
 /**
+ * AES解密
+ *
  * @author AE86
  * @version 1.0.0
  * @date 2019/10/8 23:04
  */
-public class AesDecryptHandler implements Handler {
-}
+public class AesDecryptHandler extends AbstractHandler {
+
+    @Override
+    public Object convert(String args, Object value) throws Exception {
+        return AESEncyptUtil.decrypt(String.valueOf(value), args);
+    }
+
+}

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

@@ -1,11 +1,19 @@
 package org.dbsyncer.parser.convert.handler;
 
-import org.dbsyncer.parser.convert.Handler;
+import org.dbsyncer.common.util.AESEncyptUtil;
+import org.dbsyncer.parser.convert.AbstractHandler;
 
 /**
+ * AES加密
+ *
  * @author AE86
  * @version 1.0.0
  * @date 2019/10/8 23:04
  */
-public class AesEncryptHandler implements Handler {
-}
+public class AesEncryptHandler extends AbstractHandler {
+
+    @Override
+    public Object convert(String args, Object value) throws Exception {
+        return AESEncyptUtil.encrypt(String.valueOf(value), args);
+    }
+}

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

@@ -3,9 +3,19 @@ package org.dbsyncer.parser.convert.handler;
 import org.dbsyncer.parser.convert.Handler;
 
 /**
+ * 后面追加
+ *
  * @author AE86
  * @version 1.0.0
  * @date 2019/10/8 23:04
  */
 public class AppendHandler implements Handler {
-}
+
+    @Override
+    public Object handle(String args, Object value) {
+        if (null == value) {
+            return args;
+        }
+        return new StringBuilder().append(value).append(args).toString();
+    }
+}

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

@@ -8,4 +8,9 @@ import org.dbsyncer.parser.convert.Handler;
  * @date 2019/10/8 23:05
  */
 public class ClearHandler implements Handler {
-}
+
+    @Override
+    public Object handle(String args, Object value) {
+        return null;
+    }
+}

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

@@ -1,11 +1,19 @@
 package org.dbsyncer.parser.convert.handler;
 
+import org.apache.commons.lang.StringUtils;
 import org.dbsyncer.parser.convert.Handler;
 
 /**
+ * 默认值
+ *
  * @author AE86
  * @version 1.0.0
  * @date 2019/10/8 23:02
  */
 public class DefaultHandler implements Handler {
-}
+
+    @Override
+    public Object handle(String args, Object value) {
+        return null == value || StringUtils.isBlank(String.valueOf(value)) ? args : value;
+    }
+}

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

@@ -1,5 +1,6 @@
 package org.dbsyncer.parser.convert.handler;
 
+import org.apache.commons.lang.StringUtils;
 import org.dbsyncer.parser.convert.Handler;
 
 /**
@@ -8,4 +9,9 @@ import org.dbsyncer.parser.convert.Handler;
  * @date 2019/10/8 23:03
  */
 public class DefaultTimestampHandler implements Handler {
-}
+
+    @Override
+    public Object handle(String args, Object value) {
+        return null == value || StringUtils.isBlank(String.valueOf(value)) ? System.currentTimeMillis() : value;
+    }
+}

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

@@ -3,9 +3,19 @@ package org.dbsyncer.parser.convert.handler;
 import org.dbsyncer.parser.convert.Handler;
 
 /**
+ * 前面追加
+ *
  * @author AE86
  * @version 1.0.0
  * @date 2019/10/8 23:04
  */
 public class PrependHandler implements Handler {
-}
+
+    @Override
+    public Object handle(String args, Object value) {
+        if (null == value) {
+            return args;
+        }
+        return new StringBuilder().append(args).append(value).toString();
+    }
+}

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

@@ -1,11 +1,19 @@
 package org.dbsyncer.parser.convert.handler;
 
-import org.dbsyncer.parser.convert.Handler;
+import org.apache.commons.lang.StringUtils;
+import org.dbsyncer.parser.convert.AbstractHandler;
 
 /**
+ * 去掉首字符
+ *
  * @author AE86
  * @version 1.0.0
  * @date 2019/10/8 23:05
  */
-public class RemStrFirstHandler implements Handler {
-}
+public class RemStrFirstHandler extends AbstractHandler {
+
+    @Override
+    protected Object convert(String args, Object value) {
+        return StringUtils.substring(String.valueOf(value), 1);
+    }
+}

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

@@ -1,11 +1,20 @@
 package org.dbsyncer.parser.convert.handler;
 
-import org.dbsyncer.parser.convert.Handler;
+import org.apache.commons.lang.StringUtils;
+import org.dbsyncer.parser.convert.AbstractHandler;
 
 /**
+ * 去掉尾字符
+ *
  * @author AE86
  * @version 1.0.0
  * @date 2019/10/8 23:05
  */
-public class RemStrLastHandler implements Handler {
-}
+public class RemStrLastHandler extends AbstractHandler {
+
+    @Override
+    protected Object convert(String args, Object value) {
+        String s = String.valueOf(value);
+        return StringUtils.substring(s, 0, s.length() - 1);
+    }
+}

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

@@ -1,11 +1,24 @@
 package org.dbsyncer.parser.convert.handler;
 
-import org.dbsyncer.parser.convert.Handler;
+import org.apache.commons.lang.StringUtils;
+import org.dbsyncer.parser.convert.AbstractHandler;
+import org.springframework.util.Assert;
 
 /**
+ * 替换
+ *
  * @author AE86
  * @version 1.0.0
  * @date 2019/10/8 23:03
  */
-public class ReplaceHandler implements Handler {
-}
+public class ReplaceHandler extends AbstractHandler {
+
+    @Override
+    protected Object convert(String args, Object value) {
+        Assert.hasText(args, "缺少替换参数.");
+        String[] split = StringUtils.split(args, ",");
+        String a = split[0];
+        String b = split.length == 2 ? split[1] : "";
+        return StringUtils.replace(String.valueOf(value), a, b);
+    }
+}

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

@@ -1,11 +1,19 @@
 package org.dbsyncer.parser.convert.handler;
 
-import org.dbsyncer.parser.convert.Handler;
+import org.dbsyncer.common.util.SHA1Util;
+import org.dbsyncer.parser.convert.AbstractHandler;
 
 /**
+ * SHA1加密
+ *
  * @author AE86
  * @version 1.0.0
  * @date 2019/10/8 23:04
  */
-public class Sha1Handler implements Handler {
-}
+public class Sha1Handler extends AbstractHandler {
+
+    @Override
+    protected Object convert(String args, Object value) {
+        return SHA1Util.b64_sha1(String.valueOf(value));
+    }
+}

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

@@ -1,11 +1,24 @@
 package org.dbsyncer.parser.convert.handler;
 
-import org.dbsyncer.parser.convert.Handler;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.lang.math.NumberUtils;
+import org.dbsyncer.parser.convert.AbstractHandler;
+import org.springframework.util.Assert;
 
 /**
+ * 从前面截取N个字符
+ *
  * @author AE86
  * @version 1.0.0
  * @date 2019/10/8 23:05
  */
-public class SubStrFirstHandler implements Handler {
-}
+public class SubStrFirstHandler extends AbstractHandler {
+
+    @Override
+    protected Object convert(String args, Object value) {
+        Assert.isTrue(NumberUtils.isNumber(args), "参数必须为正整数.");
+        String s = String.valueOf(value);
+        int size = NumberUtils.toInt(args);
+        return StringUtils.substring(s, 0, size);
+    }
+}

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

@@ -1,11 +1,25 @@
 package org.dbsyncer.parser.convert.handler;
 
-import org.dbsyncer.parser.convert.Handler;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.lang.math.NumberUtils;
+import org.dbsyncer.parser.convert.AbstractHandler;
+import org.springframework.util.Assert;
 
 /**
+ * 从后面截取N个字符
+ *
  * @author AE86
  * @version 1.0.0
  * @date 2019/10/8 23:05
  */
-public class SubStrLastHandler implements Handler {
-}
+public class SubStrLastHandler extends AbstractHandler {
+
+    @Override
+    protected Object convert(String args, Object value) {
+        Assert.isTrue(NumberUtils.isNumber(args), "参数必须为正整数.");
+        String s = String.valueOf(value);
+        int size = NumberUtils.toInt(args);
+        int length = s.length();
+        return StringUtils.substring(s, length - size, length);
+    }
+}

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

@@ -1,11 +1,19 @@
 package org.dbsyncer.parser.convert.handler;
 
+import org.dbsyncer.common.util.UUIDUtil;
 import org.dbsyncer.parser.convert.Handler;
 
 /**
+ * UUID
+ *
  * @author AE86
  * @version 1.0.0
  * @date 2019/10/8 23:05
  */
 public class UUIDHandler implements Handler {
-}
+
+    @Override
+    public Object handle(String args, Object value) {
+        return UUIDUtil.getUUID();
+    }
+}

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

@@ -19,9 +19,9 @@ public enum ConvertEnum {
      */
     DEFAULT("DEFAULT", "默认值", 1, new DefaultHandler()),
     /**
-     * 系统时间
+     * 默认系统时间
      */
-    SYSTEM_TIMESTAMP("SYSTEM_TIMESTAMP", "系统时间", 0, new DefaultTimestampHandler()),
+    DEFAULT_SYSTEM_TIMESTAMP("DEFAULT_SYSTEM_TIMESTAMP", "系统时间", 0, new DefaultTimestampHandler()),
     /**
      * 替换
      */

dbsyncer-parser/src/main/java/org/dbsyncer/parser/util/ConvertUtil.java

@@ -2,6 +2,7 @@ package org.dbsyncer.parser.util;
 
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.parser.convert.Convert;
+import org.dbsyncer.parser.enums.ConvertEnum;
 
 import java.util.List;
 import java.util.Map;
@@ -19,7 +20,25 @@ public abstract class ConvertUtil {
      */
     public static void convert(List<Convert> convert, List<Map<String, Object>> data) {
         if (!CollectionUtils.isEmpty(convert) && !CollectionUtils.isEmpty(data)) {
-            // TODO 参数转换
+            // 并行流计算
+            final int size = convert.size();
+            data.parallelStream().forEach(row -> {
+                // 替换row值, 复用堆栈地址，减少开销
+                Convert c = null;
+                String name = null;
+                String code = null;
+                String args = null;
+                Object value = null;
+                for (int i = 0; i < size; i++) {
+                    c = convert.get(i);
+                    name = c.getName();
+                    code = c.getConvertCode();
+                    args = c.getArgs();
+                    value = ConvertEnum.getHandler(code).handle(args, row.get(name));
+
+                    row.put(name, value);
+                }
+            });
         }
     }
 

dbsyncer-parser/src/main/resources/Mapping.json

@@ -38,9 +38,10 @@
   ],
   "convert": [
     {
-      "name": "NAME",
-      "convert": "REPLACE",
-      "args": "A,B"
+      "name":"NAME",
+      "convertName":"替换",
+      "convertCode":"REPLACE",
+      "args":"A,B"
     }
   ],
   "plugin": {

dbsyncer-parser/src/main/resources/TableGroup.json

@@ -65,9 +65,10 @@
   ],
   "convert": [
     {
-      "name": "NAME",
-      "convert": "REPLACE",
-      "args": "A,B"
+      "name":"NAME",
+      "convertName":"替换",
+      "convertCode":"REPLACE",
+      "args":"A,B"
     }
   ],
   "plugin": {