支持手动重试 https://gitee.com/ghi/dbsyncer/issues/I5IY1B

dbsyncer-biz/src/main/java/org/dbsyncer/biz/MonitorService.java

@@ -56,6 +56,14 @@ public interface MonitorService {
      */
     MessageVo getMessageVo(String metaId, String messageId);
 
+    /**
+     * 手动同步单条数据
+     *
+     * @param params
+     * @return
+     */
+    String sync(Map<String, String> params);
+
     /**
      * 清空驱动同步数据
      *

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

@@ -10,6 +10,7 @@ import org.dbsyncer.biz.metric.impl.GCMetricDetailFormatter;
 import org.dbsyncer.biz.metric.impl.MemoryMetricDetailFormatter;
 import org.dbsyncer.biz.metric.impl.ValueMetricDetailFormatter;
 import org.dbsyncer.biz.vo.AppReportMetricVo;
+import org.dbsyncer.biz.vo.BinlogColumnVo;
 import org.dbsyncer.biz.vo.DataVo;
 import org.dbsyncer.biz.vo.LogVo;
 import org.dbsyncer.biz.vo.MessageVo;
@@ -18,6 +19,7 @@ import org.dbsyncer.biz.vo.MetricResponseVo;
 import org.dbsyncer.cache.CacheService;
 import org.dbsyncer.common.model.Paging;
 import org.dbsyncer.common.util.CollectionUtils;
+import org.dbsyncer.common.util.DateFormatUtil;
 import org.dbsyncer.common.util.JsonUtil;
 import org.dbsyncer.common.util.NumberUtil;
 import org.dbsyncer.common.util.StringUtil;
@@ -30,10 +32,12 @@ import org.dbsyncer.monitor.enums.ThreadPoolMetricEnum;
 import org.dbsyncer.monitor.model.AppReportMetric;
 import org.dbsyncer.monitor.model.MetricResponse;
 import org.dbsyncer.parser.enums.ModelEnum;
+import org.dbsyncer.parser.flush.BufferActuator;
 import org.dbsyncer.parser.model.Mapping;
 import org.dbsyncer.parser.model.Meta;
 import org.dbsyncer.parser.model.Picker;
 import org.dbsyncer.parser.model.TableGroup;
+import org.dbsyncer.parser.model.WriterRequest;
 import org.dbsyncer.storage.binlog.proto.BinlogMap;
 import org.dbsyncer.storage.constant.ConfigConstant;
 import org.dbsyncer.storage.enums.StorageDataStatusEnum;
@@ -71,6 +75,9 @@ public class MonitorServiceImpl implements MonitorService {
     @Autowired
     private CacheService cacheService;
 
+    @Autowired
+    private BufferActuator writerBufferActuator;
+
     private Map<String, MetricDetailFormatter> metricDetailFormatterMap = new LinkedHashMap<>();
 
     @PostConstruct
@@ -133,7 +140,7 @@ public class MonitorServiceImpl implements MonitorService {
         for (Map row : data) {
             try {
                 DataVo dataVo = convert2Vo(row, DataVo.class);
-                Map binlogData = getBinlogData(row);
+                Map binlogData = getBinlogData(row, true);
                 dataVo.setJson(JsonUtil.objToJson(binlogData));
                 list.add(dataVo);
             } catch (Exception e) {
@@ -152,17 +159,50 @@ public class MonitorServiceImpl implements MonitorService {
         MessageVo messageVo = new MessageVo();
         try {
             Map row = monitor.getData(metaId, messageId);
-            Map binlogData = getBinlogData(row);
+            Map binlogData = getBinlogData(row, true);
             String tableGroupId = (String) row.get(ConfigConstant.DATA_TABLE_GROUP_ID);
             TableGroup tableGroup = monitor.getTableGroup(tableGroupId);
-            messageVo.setTableGroup(tableGroup);
-            messageVo.setRow(binlogData);
+            messageVo.setSourceTableName(tableGroup.getSourceTable().getName());
+            messageVo.setTargetTableName(tableGroup.getTargetTable().getName());
+            messageVo.setId(messageId);
+
+            if (!CollectionUtils.isEmpty(binlogData)) {
+                Map<String, String> columnMap = tableGroup.getTargetTable().getColumn().stream().collect(Collectors.toMap(Field::getName, Field::getTypeName));
+                List<BinlogColumnVo> columns = new ArrayList<>();
+                binlogData.forEach((k, v) -> columns.add(new BinlogColumnVo((String) k, v, columnMap.get(k))));
+                messageVo.setColumns(columns);
+            }
         } catch (Exception e) {
             logger.error(e.getLocalizedMessage());
         }
         return messageVo;
     }
 
+    @Override
+    public String sync(Map<String, String> params) {
+        String metaId = params.get("metaId");
+        String messageId = params.get("messageId");
+        Assert.hasText(metaId, "The metaId is null.");
+        Assert.hasText(messageId, "The messageId is null.");
+
+        try {
+            Map row = monitor.getData(metaId, messageId);
+            Map binlogData = getBinlogData(row, false);
+            String tableGroupId = (String) row.get(ConfigConstant.DATA_TABLE_GROUP_ID);
+            String event = (String) row.get(ConfigConstant.DATA_EVENT);
+            // 有修改同步值
+            String retryDataParams = params.get("retryDataParams");
+            if (StringUtil.isNotBlank(retryDataParams)) {
+                JsonUtil.parseObject(retryDataParams).getInnerMap().forEach((k, v) -> binlogData.put(k, convertValue(binlogData.get(k), (String) v)));
+            }
+            writerBufferActuator.offer(new WriterRequest(tableGroupId, event, binlogData));
+            monitor.removeData(metaId, messageId);
+        } catch (Exception e) {
+            logger.error(e.getLocalizedMessage());
+        }
+        return messageId;
+    }
+
     @Override
     public String clearData(String id) {
         Assert.hasText(id, "驱动不存在.");
@@ -208,7 +248,7 @@ public class MonitorServiceImpl implements MonitorService {
         return vo;
     }
 
-    private Map getBinlogData(Map row) throws InvalidProtocolBufferException {
+    private Map getBinlogData(Map row, boolean prettyBytes) throws InvalidProtocolBufferException {
         String tableGroupId = (String) row.get(ConfigConstant.DATA_TABLE_GROUP_ID);
         byte[] bytes = (byte[]) row.get(ConfigConstant.BINLOG_DATA);
         BinlogMap message = BinlogMap.parseFrom(bytes);
@@ -224,14 +264,16 @@ public class MonitorServiceImpl implements MonitorService {
             if (fieldMap.containsKey(k)) {
                 Object val = BinlogMessageUtil.deserializeValue(fieldMap.get(k).getType(), v);
                 // 处理二进制对象显示
-                if (null != val && val instanceof byte[]) {
-                    byte[] b = (byte[]) val;
-                    if (b.length > 128) {
-                        map.put(k, String.format("bytes[%d]", b.length));
+                if (prettyBytes) {
+                    if (null != val && val instanceof byte[]) {
+                        byte[] b = (byte[]) val;
+                        if (b.length > 128) {
+                            map.put(k, String.format("bytes[%d]", b.length));
+                            return;
+                        }
+                        map.put(k, Arrays.toString(b));
                         return;
                     }
-                    map.put(k, Arrays.toString(b));
-                    return;
                 }
                 map.put(k, val);
             }
@@ -239,6 +281,51 @@ public class MonitorServiceImpl implements MonitorService {
         return map;
     }
 
+    private Object convertValue(Object oldValue, String newValue) {
+        if (oldValue == null) {
+            return newValue;
+        }
+
+        Object newVal = null;
+        String type = oldValue.getClass().getName();
+        switch (type) {
+            case "java.sql.Date":
+                newVal = DateFormatUtil.stringToDate(newValue);
+                break;
+            case "java.sql.Timestamp":
+                newVal = DateFormatUtil.stringToTimestamp(newValue);
+                break;
+            case "java.sql.Integer":
+            case "java.sql.Short":
+                newVal = NumberUtil.toInt(newValue);
+                break;
+            case "java.lang.Float":
+                newVal = Float.valueOf(newValue);
+                break;
+            case "[B":
+                newVal = stringToBytes(newValue);
+                break;
+            default:
+                newVal = newValue;
+        }
+
+        return newVal;
+    }
+
+    private byte[] stringToBytes(String s) {
+        byte[] b = null;
+        if (s.startsWith("[") && s.endsWith("]")) {
+            s = StringUtil.substring(s, 1, s.length() - 1);
+            String[] split = StringUtil.split(s, ",");
+            int length = split.length;
+            b = new byte[length];
+            for (int i = 0; i < length; i++) {
+                b[i] = Byte.valueOf(split[i].trim());
+            }
+        }
+        return b;
+    }
+
     private MetaVo convertMeta2Vo(Meta meta) {
         Mapping mapping = monitor.getMapping(meta.getMappingId());
         Assert.notNull(mapping, "驱动不存在.");

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

@@ -0,0 +1,35 @@
+package org.dbsyncer.biz.vo;
+
+public class BinlogColumnVo {
+
+    private String key;
+
+    private Object value;
+
+    private String keyType;
+
+    private String valueType;
+
+    public BinlogColumnVo(String key, Object value, String keyType) {
+        this.key = key;
+        this.value = value;
+        this.keyType = keyType;
+        this.valueType = value == null ? "" : value.getClass().getName();
+    }
+
+    public String getKey() {
+        return key;
+    }
+
+    public Object getValue() {
+        return value;
+    }
+
+    public String getKeyType() {
+        return keyType;
+    }
+
+    public String getValueType() {
+        return valueType;
+    }
+}

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

@@ -1,28 +1,46 @@
 package org.dbsyncer.biz.vo;
 
-import org.dbsyncer.parser.model.TableGroup;
-
-import java.util.Map;
+import java.util.List;
 
 public class MessageVo {
 
-    private TableGroup tableGroup;
+    private String id;
+
+    private String sourceTableName;
+
+    private String targetTableName;
+
+    private List<BinlogColumnVo> columns;
+
+    public String getId() {
+        return id;
+    }
 
-    private Map row;
+    public void setId(String id) {
+        this.id = id;
+    }
+
+    public String getSourceTableName() {
+        return sourceTableName;
+    }
+
+    public void setSourceTableName(String sourceTableName) {
+        this.sourceTableName = sourceTableName;
+    }
 
-    public TableGroup getTableGroup() {
-        return tableGroup;
+    public String getTargetTableName() {
+        return targetTableName;
     }
 
-    public void setTableGroup(TableGroup tableGroup) {
-        this.tableGroup = tableGroup;
+    public void setTargetTableName(String targetTableName) {
+        this.targetTableName = targetTableName;
     }
 
-    public Map getRow() {
-        return row;
+    public List<BinlogColumnVo> getColumns() {
+        return columns;
     }
 
-    public void setRow(Map row) {
-        this.row = row;
+    public void setColumns(List<BinlogColumnVo> columns) {
+        this.columns = columns;
     }
 }

dbsyncer-common/src/main/java/org/dbsyncer/common/util/DateFormatUtil.java

@@ -8,8 +8,18 @@ import java.sql.Timestamp;
 import java.text.DateFormat;
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
-import java.time.*;
-import java.time.format.*;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.OffsetDateTime;
+import java.time.OffsetTime;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeFormatterBuilder;
+import java.time.format.DateTimeParseException;
+import java.time.format.SignStyle;
+import java.time.format.TextStyle;
 import java.time.temporal.ChronoField;
 import java.time.temporal.TemporalAccessor;
 
@@ -106,6 +116,12 @@ public abstract class DateFormatUtil {
                 return Timestamp.valueOf(LocalDateTime.from(CHINESE_STANDARD_TIME_FORMATTER.parse(s)));
             }
 
+            // 2020-07-12 00:00:00.0
+            if (s.length() == 21) {
+                s = s.substring(0, s.lastIndexOf("."));
+                return Timestamp.valueOf(LocalDateTime.from(CHINESE_STANDARD_TIME_FORMATTER.parse(s)));
+            }
+
             // 2022-07-21T05:35:34.000+0800
             if (s.length() == 28) {
                 return stringToTimestamp(s, GMT_FORMATTER);

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

@@ -109,6 +109,8 @@ public interface Manager extends Executor {
     // Data
     Paging queryData(Query query);
 
+    void removeData(String metaId, String messageId);
+
     void clearData(String metaId);
 
     // Log

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

@@ -225,6 +225,11 @@ public class ManagerFactory implements Manager, ApplicationListener<ClosedEvent>
         return storageService.query(query);
     }
 
+    @Override
+    public void removeData(String metaId, String messageId) {
+        storageService.remove(StorageEnum.DATA, metaId, messageId);
+    }
+
     @Override
     public void clearData(String metaId) {
         Meta meta = getMeta(metaId);

dbsyncer-monitor/src/main/java/org/dbsyncer/monitor/Monitor.java

@@ -31,6 +31,8 @@ public interface Monitor {
 
     Map getData(String metaId, String messageId);
 
+    void removeData(String metaId, String messageId);
+
     void clearData(String metaId);
 
     Paging queryLog(int pageNum, int pageSize, String json);

dbsyncer-monitor/src/main/java/org/dbsyncer/monitor/MonitorFactory.java

@@ -138,6 +138,11 @@ public class MonitorFactory implements Monitor, ScheduledTaskJob {
         return Collections.EMPTY_MAP;
     }
 
+    @Override
+    public void removeData(String metaId, String messageId) {
+        manager.removeData(metaId, messageId);
+    }
+
     @Override
     public void clearData(String metaId) {
         manager.clearData(metaId);

dbsyncer-web/src/main/java/org/dbsyncer/web/controller/monitor/MonitorController.java

@@ -125,7 +125,8 @@ public class MonitorController extends BaseController {
     @ResponseBody
     public RestResult sync(HttpServletRequest request) {
         try {
-            return RestResult.restSuccess("ok");
+            Map<String, String> params = getParams(request);
+            return RestResult.restSuccess(monitorService.sync(params));
         } catch (Exception e) {
             logger.error(e.getLocalizedMessage(), e.getClass());
             return RestResult.restFail(e.getMessage());

dbsyncer-web/src/main/resources/public/monitor/monitor.html

@@ -142,7 +142,7 @@
                                     <td th:text="${#dates.format(d?.createTime, 'yyyy-MM-dd HH:mm:ss')}" />
                                     <td>
                                         <a th:json="${d?.json}" href="javascript:;" class="label label-info queryData">查看数据</a>
-                                        <a th:id="${d?.id}" href="javascript:;" class="label label-warning retryData">重试</a>
+                                        <a th:if="${d?.success == 0}" th:id="${d?.id}" href="javascript:;" class="label label-warning retryData">重试</a>
                                         <div class="hidden" th:text="${d?.json}"></div>
                                     </td>
                                 </tr>

dbsyncer-web/src/main/resources/public/monitor/retry.html

@@ -9,13 +9,15 @@ xmlns:th="http://www.thymeleaf.org" lang="zh-CN">
       <div class="row text-center">
         <div class="page-header">
           <h3>重试[[${meta?.mappingName}]]</h3>
-          <input type="hidden" id="metaId" th:value="${meta?.id}">
+          <input type="hidden" name="metaId" id="metaId" th:value="${meta?.id}"/>
+          <input type="hidden" name="messageId" th:value="${message?.id}"/>
+          <input type="hidden" id="retryDataParams" name="retryDataParams"/>
         </div>
       </div>
 
       <!-- 操作 -->
       <div class="form-group">
-        <div class="col-md-10"><h3>[[${message?.tableGroup?.sourceTable?.name}]] <span class="fa fa-angle-double-right"></span> [[${message?.tableGroup?.targetTable?.name}]]</h3></div>
+        <div class="col-md-10"><h4>[[${message?.sourceTableName}]] <span class="fa fa-angle-double-right"></span> [[${message?.targetTableName}]]</h4></div>
         <div class="col-md-2 text-right">
           <button id="retryDataSubmitBtn" type="button" class="btn btn-primary">
             <span class="fa fa-refresh"></span>执行
@@ -35,14 +37,18 @@ xmlns:th="http://www.thymeleaf.org" lang="zh-CN">
             <tr>
               <th></th>
               <th>目标表字段</th>
-              <th>目标表值</th>
+              <th>值类型</th>
+              <th>值</th>
+              <th>操作</th>
             </tr>
             </thead>
-            <tbody id="pluginList">
-            <tr th:each="m,state : ${message?.row}">
+            <tbody>
+            <tr th:each="c,state : ${message?.columns}">
               <td th:text="${state.index}+1"></td>
-              <td th:text="${state?.current?.key}"/>
-              <td th:text="${state?.current?.value}"/>
+              <td th:text="${c?.key + '(' + c?.keyType + ')'}"/>
+              <td th:text="${c?.valueType}"/>
+              <td th:text="${c?.value}"/>
+              <td><i th:type="${c?.key}" th:title="修改" class="fa fa-edit well-sign-green dbsyncer_pointer retryDataModify"></i></td>
             </tr>
             </tbody>
           </table>

dbsyncer-web/src/main/resources/static/js/monitor/index.js

@@ -198,7 +198,9 @@ function showData($dataList, arr, append) {
             html += '<td>' + (arr[i].success ? '<span class="label label-success">成功</span>' : '<span class="label label-warning">失败</span>') + '</td>';
             html += '<td style="max-width:100px;" class="dbsyncer_over_hidden"><a href="javascript:;" class="dbsyncer_pointer queryError">' + arr[i].error + '</a></td>';
             html += '<td>' + formatDate(arr[i].createTime) + '</td>';
-            html += '<td><a href="javascript:;" class="label label-info queryData">查看数据</a>&nbsp;<a id="' + arr[i].id + '" href="javascript:;" class="label label-warning retryData">重试</a><div class="hidden">' + arr[i].json + '</div></td>';
+            html += '<td><div class="hidden">' + arr[i].json + '</div><a href="javascript:;" class="label label-info queryData">查看数据</a>&nbsp;';
+            html += (arr[i].success ? '' : '<a id="' + arr[i].id + '" href="javascript:;" class="label label-warning retryData">重试</a>');
+            html += '</td>';
             html += '</tr>';
         }
     }

dbsyncer-web/src/main/resources/static/js/monitor/retry.js

@@ -13,6 +13,42 @@ function backMonitorPage() {
     doLoader('/monitor?id=' + $("#metaId").val());
 }
 
+// 修改增量数据
+function bindRetryDataModifyClick() {
+    $(".retryDataModify").click(function () {
+        let $key = $(this).attr("type");
+        let $tr = $(this).parent().parent();
+        let $value = $tr.find("td:eq(3)");
+        let tmp = $value.text();
+        $value.text("");
+        $value.append("<input type='text'/>");
+        let $input = $value.find("input");
+        $input.focus().val(tmp);
+        $input.blur(function () {
+            $value.text($(this).val());
+            if (tmp != $(this).val()) {
+                createRetryDataParams($key, $(this).val());
+                $tr.addClass("success");
+                $tr.attr("title", "已修改");
+            }
+            $input.unbind();
+        });
+    })
+}
+
+// 生成修改参数
+function createRetryDataParams($key, $value) {
+    let $params = $("#retryDataParams");
+    let val = $params.val();
+    let jsonObj = {};
+    if(!isBlank(val)){
+        jsonObj = $.parseJSON(val);
+    }
+
+    jsonObj[$key] = $value;
+    $params.val(JSON.stringify(jsonObj));
+}
+
 $(function () {
     //保存
     $("#retryDataSubmitBtn").click(function () {
@@ -25,4 +61,6 @@ $(function () {
     $("#retryDataBackBtn").click(function () {
         backMonitorPage();
     });
+
+    bindRetryDataModifyClick();
 })