gen proto

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

@@ -5,10 +5,12 @@ import org.dbsyncer.common.event.RowChangedEvent;
 import org.dbsyncer.common.model.Result;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.JsonUtil;
-import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.ConnectorFactory;
 import org.dbsyncer.connector.ConnectorMapper;
-import org.dbsyncer.connector.config.*;
+import org.dbsyncer.connector.config.CommandConfig;
+import org.dbsyncer.connector.config.ConnectorConfig;
+import org.dbsyncer.connector.config.ReaderConfig;
+import org.dbsyncer.connector.config.WriterBatchConfig;
 import org.dbsyncer.connector.constant.ConnectorConstant;
 import org.dbsyncer.connector.enums.ConnectorEnum;
 import org.dbsyncer.connector.enums.FilterEnum;
@@ -20,12 +22,11 @@ import org.dbsyncer.listener.enums.QuartzFilterEnum;
 import org.dbsyncer.parser.enums.ConvertEnum;
 import org.dbsyncer.parser.enums.ParserEnum;
 import org.dbsyncer.parser.event.FullRefreshEvent;
-import org.dbsyncer.parser.flush.BufferActuator;
-import org.dbsyncer.parser.flush.FlushStrategy;
-import org.dbsyncer.parser.model.WriterRequest;
 import org.dbsyncer.parser.logger.LogService;
 import org.dbsyncer.parser.logger.LogType;
 import org.dbsyncer.parser.model.*;
+import org.dbsyncer.parser.strategy.FlushStrategy;
+import org.dbsyncer.parser.strategy.ParserStrategy;
 import org.dbsyncer.parser.util.ConvertUtil;
 import org.dbsyncer.parser.util.PickerUtil;
 import org.dbsyncer.plugin.PluginFactory;
@@ -81,7 +82,7 @@ public class ParserFactory implements Parser {
     private ApplicationContext applicationContext;
 
     @Autowired
-    private BufferActuator writerBufferActuator;
+    private ParserStrategy parserStrategy;
 
     @Override
     public ConnectorMapper connect(ConnectorConfig config) {
@@ -300,20 +301,7 @@ public class ParserFactory implements Parser {
         logger.info("Table[{}] {}, before:{}, after:{}", event.getSourceTableName(), event.getEvent(),
                 event.getBefore(), event.getAfter());
 
-        // 1、获取映射字段
-        final String eventName = event.getEvent();
-        Map<String, Object> data = StringUtil.equals(ConnectorConstant.OPERTION_DELETE, eventName) ? event.getBefore() : event.getAfter();
-        Picker picker = new Picker(tableGroup.getFieldMapping());
-        Map target = picker.pickData(data);
-
-        // 2、参数转换
-        ConvertUtil.convert(tableGroup.getConvert(), target);
-
-        // 3、插件转换
-        pluginFactory.convert(tableGroup.getPlugin(), eventName, data, target);
-
-        // 4、写入缓冲执行器
-        writerBufferActuator.offer(new WriterRequest(tableGroup.getId(), target, mapping.getMetaId(), mapping.getTargetConnectorId(), event.getSourceTableName(), event.getTargetTableName(), eventName, picker.getTargetFields(), tableGroup.getCommand()));
+        parserStrategy.execute(mapping, tableGroup, event);
     }
 
     /**

dbsyncer-parser/src/main/java/org/dbsyncer/parser/config/ParserFlushStrategyConfiguration.java → dbsyncer-parser/src/main/java/org/dbsyncer/parser/config/ParserStrategyConfiguration.java

@@ -1,7 +1,9 @@
 package org.dbsyncer.parser.config;
 
-import org.dbsyncer.parser.flush.FlushStrategy;
-import org.dbsyncer.parser.flush.impl.DisableFullFlushStrategy;
+import org.dbsyncer.parser.strategy.FlushStrategy;
+import org.dbsyncer.parser.strategy.ParserStrategy;
+import org.dbsyncer.parser.strategy.impl.DisableFullFlushStrategy;
+import org.dbsyncer.parser.strategy.impl.DisableWriterBufferActuatorStrategy;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -12,7 +14,7 @@ import org.springframework.context.annotation.Configuration;
  * @date 2021/11/18 21:36
  */
 @Configuration
-public class ParserFlushStrategyConfiguration {
+public class ParserStrategyConfiguration {
 
     @Bean
     @ConditionalOnMissingBean
@@ -20,4 +22,10 @@ public class ParserFlushStrategyConfiguration {
         return new DisableFullFlushStrategy();
     }
 
+    @Bean
+    @ConditionalOnMissingBean
+    public ParserStrategy parserStrategy() {
+        return new DisableWriterBufferActuatorStrategy();
+    }
+
 }

dbsyncer-parser/src/main/java/org/dbsyncer/parser/flush/AbstractFlushStrategy.java

@@ -5,6 +5,7 @@ import org.dbsyncer.common.model.Result;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.parser.model.Meta;
+import org.dbsyncer.parser.strategy.FlushStrategy;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.util.Assert;
 

dbsyncer-parser/src/main/java/org/dbsyncer/parser/flush/impl/WriterBufferActuator.java

@@ -7,7 +7,7 @@ import org.dbsyncer.connector.ConnectorMapper;
 import org.dbsyncer.connector.config.ConnectorConfig;
 import org.dbsyncer.parser.ParserFactory;
 import org.dbsyncer.parser.flush.AbstractBufferActuator;
-import org.dbsyncer.parser.flush.FlushStrategy;
+import org.dbsyncer.parser.strategy.FlushStrategy;
 import org.dbsyncer.parser.model.WriterRequest;
 import org.dbsyncer.parser.model.WriterResponse;
 import org.dbsyncer.parser.model.BatchWriter;

dbsyncer-parser/src/main/java/org/dbsyncer/parser/flush/FlushStrategy.java → dbsyncer-parser/src/main/java/org/dbsyncer/parser/strategy/FlushStrategy.java

@@ -1,4 +1,4 @@
-package org.dbsyncer.parser.flush;
+package org.dbsyncer.parser.strategy;
 
 import org.dbsyncer.common.model.Result;
 

dbsyncer-parser/src/main/java/org/dbsyncer/parser/strategy/ParserStrategy.java

@@ -0,0 +1,10 @@
+package org.dbsyncer.parser.strategy;
+
+import org.dbsyncer.common.event.RowChangedEvent;
+import org.dbsyncer.parser.model.Mapping;
+import org.dbsyncer.parser.model.TableGroup;
+
+public interface ParserStrategy {
+
+    void execute(Mapping mapping, TableGroup tableGroup, RowChangedEvent event);
+}

dbsyncer-parser/src/main/java/org/dbsyncer/parser/flush/impl/DisableFullFlushStrategy.java → dbsyncer-parser/src/main/java/org/dbsyncer/parser/strategy/impl/DisableFullFlushStrategy.java

@@ -1,4 +1,4 @@
-package org.dbsyncer.parser.flush.impl;
+package org.dbsyncer.parser.strategy.impl;
 
 import org.dbsyncer.common.model.Result;
 import org.dbsyncer.common.util.CollectionUtils;

dbsyncer-parser/src/main/java/org/dbsyncer/parser/strategy/impl/DisableWriterBufferActuatorStrategy.java

@@ -0,0 +1,63 @@
+package org.dbsyncer.parser.strategy.impl;
+
+import com.google.protobuf.ByteString;
+import org.dbsyncer.common.event.RowChangedEvent;
+import org.dbsyncer.common.util.StringUtil;
+import org.dbsyncer.connector.constant.ConnectorConstant;
+import org.dbsyncer.parser.flush.BufferActuator;
+import org.dbsyncer.parser.model.Mapping;
+import org.dbsyncer.parser.model.TableGroup;
+import org.dbsyncer.parser.model.WriterRequest;
+import org.dbsyncer.parser.strategy.ParserStrategy;
+import org.dbsyncer.storage.binlog.AbstractBinlogRecorder;
+import org.dbsyncer.storage.binlog.proto.BinlogMessage;
+import org.dbsyncer.storage.binlog.proto.Data;
+import org.dbsyncer.storage.binlog.proto.EventEnum;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import java.util.Map;
+import java.util.Queue;
+
+public final class DisableWriterBufferActuatorStrategy extends AbstractBinlogRecorder<WriterRequest> implements ParserStrategy {
+
+    private final Logger logger = LoggerFactory.getLogger(getClass());
+
+    @Autowired
+    private BufferActuator writerBufferActuator;
+
+    @Override
+    public void execute(Mapping mapping, TableGroup tableGroup, RowChangedEvent event) {
+        try {
+            EventEnum eventEnum = EventEnum.valueOf(event.getEvent());
+            Map<String, Object> data = StringUtil.equals(ConnectorConstant.OPERTION_DELETE, eventEnum.name()) ? event.getBefore() : event.getAfter();
+            BinlogMessage.Builder builder = BinlogMessage.newBuilder()
+                    .setTableGroupId(tableGroup.getId())
+                    .setEvent(eventEnum);
+            data.forEach((k, v) -> {
+                if (null != v && v instanceof String) {
+                    builder.addData(Data.newBuilder().putRow(k, ByteString.copyFromUtf8((String) v)));
+                }
+            });
+            flush(builder.build());
+        } catch (Exception e) {
+            logger.error(e.getMessage());
+        }
+    }
+
+    @Override
+    protected String getTaskName() {
+        return "WriterBinlog";
+    }
+
+    @Override
+    protected Queue getQueue() {
+        return writerBufferActuator.getQueue();
+    }
+
+    @Override
+    protected WriterRequest deserialize(BinlogMessage message) {
+        return null;
+    }
+}

dbsyncer-parser/src/main/java/org/dbsyncer/parser/flush/impl/EnableFlushStrategy.java → dbsyncer-parser/src/main/java/org/dbsyncer/parser/strategy/impl/EnableFlushStrategy.java

@@ -1,4 +1,4 @@
-package org.dbsyncer.parser.flush.impl;
+package org.dbsyncer.parser.strategy.impl;
 
 import org.dbsyncer.parser.flush.AbstractFlushStrategy;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;

dbsyncer-parser/src/main/java/org/dbsyncer/parser/strategy/impl/EnableWriterBufferActuatorStrategy.java

@@ -0,0 +1,48 @@
+package org.dbsyncer.parser.strategy.impl;
+
+import org.dbsyncer.common.event.RowChangedEvent;
+import org.dbsyncer.common.util.StringUtil;
+import org.dbsyncer.connector.constant.ConnectorConstant;
+import org.dbsyncer.parser.flush.BufferActuator;
+import org.dbsyncer.parser.model.Mapping;
+import org.dbsyncer.parser.model.Picker;
+import org.dbsyncer.parser.model.TableGroup;
+import org.dbsyncer.parser.model.WriterRequest;
+import org.dbsyncer.parser.strategy.ParserStrategy;
+import org.dbsyncer.parser.util.ConvertUtil;
+import org.dbsyncer.plugin.PluginFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Component;
+
+import java.util.Map;
+
+@Component
+@ConditionalOnProperty(value = "dbsyncer.parser.writer.buffer.actuator.enabled", havingValue = "true")
+public final class EnableWriterBufferActuatorStrategy implements ParserStrategy {
+
+    @Autowired
+    private PluginFactory pluginFactory;
+
+    @Autowired
+    private BufferActuator writerBufferActuator;
+
+    @Override
+    public void execute(Mapping mapping, TableGroup tableGroup, RowChangedEvent event) {
+        // 1、获取映射字段
+        final String eventName = event.getEvent();
+        Map<String, Object> data = StringUtil.equals(ConnectorConstant.OPERTION_DELETE, eventName) ? event.getBefore() : event.getAfter();
+        Picker picker = new Picker(tableGroup.getFieldMapping());
+        Map target = picker.pickData(data);
+
+        // 2、参数转换
+        ConvertUtil.convert(tableGroup.getConvert(), target);
+
+        // 3、插件转换
+        pluginFactory.convert(tableGroup.getPlugin(), eventName, data, target);
+
+        // 4、写入缓冲执行器
+        writerBufferActuator.offer(new WriterRequest(tableGroup.getId(), target, mapping.getMetaId(), mapping.getTargetConnectorId(), event.getSourceTableName(), event.getTargetTableName(), eventName, picker.getTargetFields(), tableGroup.getCommand()));
+    }
+
+}

dbsyncer-storage/src/main/java/org/dbsyncer/storage/binlog/AbstractBinlogRecorder.java

@@ -3,11 +3,14 @@ package org.dbsyncer.storage.binlog;
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.io.IOUtils;
 import org.dbsyncer.common.file.BufferedRandomAccessFile;
+import org.dbsyncer.common.scheduled.ScheduledTaskJob;
+import org.dbsyncer.common.scheduled.ScheduledTaskService;
 import org.dbsyncer.common.util.JsonUtil;
 import org.dbsyncer.storage.binlog.proto.BinlogMessage;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.util.Assert;
 
 import javax.annotation.PostConstruct;
@@ -15,17 +18,19 @@ import java.io.*;
 import java.nio.charset.Charset;
 import java.util.Arrays;
 import java.util.Queue;
-import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * @author AE86
  * @version 1.0.0
  * @date 2022/6/8 0:53
  */
-public abstract class AbstractBinlogRecorder implements BinlogRecorder, DisposableBean {
+public abstract class AbstractBinlogRecorder<Message> implements BinlogRecorder, ScheduledTaskJob, DisposableBean {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
+    @Autowired
+    private ScheduledTaskService scheduledTaskService;
+
     private static final long BINLOG_MAX_SIZE = 512 * 1024 * 1024;
 
     private static final int BINLOG_EXPIRE_DAYS = 7;
@@ -40,9 +45,7 @@ public abstract class AbstractBinlogRecorder implements BinlogRecorder, Disposab
 
     private static final String BINLOG_CONFIG = BINLOG + ".config";
 
-    private static final int MAX_CYCLE = 100;
-
-    private final AtomicInteger cycle = new AtomicInteger();
+    private static final long PERIOD = 3000;
 
     private String path;
 
@@ -60,7 +63,7 @@ public abstract class AbstractBinlogRecorder implements BinlogRecorder, Disposab
         path = new StringBuilder(System.getProperty("user.dir")).append(File.separatorChar)
                 .append("data").append(File.separatorChar)
                 .append("binlog").append(File.separatorChar)
-                .append(getClass().getSimpleName()).append(File.separatorChar)
+                .append(getTaskName()).append(File.separatorChar)
                 .toString();
         File dir = new File(path);
         if (!dir.exists()) {
@@ -68,6 +71,16 @@ public abstract class AbstractBinlogRecorder implements BinlogRecorder, Disposab
         }
 
         initPipeline();
+        scheduledTaskService.start(PERIOD, this);
+    }
+
+    /**
+     * 获取任务名称
+     *
+     * @return
+     */
+    protected String getTaskName() {
+        return getClass().getSimpleName();
     }
 
     /**
@@ -78,28 +91,25 @@ public abstract class AbstractBinlogRecorder implements BinlogRecorder, Disposab
     protected abstract Queue getQueue();
 
     /**
-     * 解析binlog
+     * 反序列化任务
+     *
+     * @param message
+     * @return
      */
-    protected void parseBinlog() {
+    protected abstract Message deserialize(BinlogMessage message);
+
+    @Override
+    public void run() {
         if (!getQueue().isEmpty()) {
             return;
         }
-        if (getQueue().isEmpty()) {
-            cycle.getAndAdd(1);
-            if (cycle.get() < MAX_CYCLE) {
-                return;
-            }
-        }
-
-        cycle.set(0);
 
         try {
             byte[] line;
             boolean hasLine = false;
             while (null != (line = pipeline.readLine())) {
-                BinlogMessage message = BinlogMessage.parseFrom(line);
-                logger.info("parse message:{}", message.toString());
-//                getQueue().offer(message);
+                deserialize(BinlogMessage.parseFrom(line));
+//                getQueue().offer(deserialize(message));
                 hasLine = true;
             }
 
@@ -113,12 +123,14 @@ public abstract class AbstractBinlogRecorder implements BinlogRecorder, Disposab
     }
 
     @Override
-    public void flushBinlog(BinlogMessage message) {
-        try {
-            out.write(message.toByteArray());
-            out.write(LINE_SEPARATOR.getBytes(DEFAULT_CHARSET));
-        } catch (IOException e) {
-            logger.error(e.getMessage());
+    public void flush(BinlogMessage message) {
+        if (null != message) {
+            try {
+                out.write(message.toByteArray());
+                out.write(LINE_SEPARATOR.getBytes(DEFAULT_CHARSET));
+            } catch (IOException e) {
+                logger.error(e.getMessage());
+            }
         }
     }
 

dbsyncer-storage/src/main/java/org/dbsyncer/storage/binlog/BinlogRecorder.java

@@ -14,6 +14,6 @@ public interface BinlogRecorder {
      *
      * @param message
      */
-    void flushBinlog(BinlogMessage message);
+    void flush(BinlogMessage message);
 
 }

dbsyncer-storage/src/main/java/org/dbsyncer/storage/binlog/proto/BinlogMessageOrBuilder.java

@@ -23,18 +23,18 @@ public interface BinlogMessageOrBuilder extends
     getTableGroupIdBytes();
 
     /**
-     * <code>.Event event = 2;</code>
+     * <code>.EventEnum event = 2;</code>
      *
      * @return The enum numeric value on the wire for event.
      */
     int getEventValue();
 
     /**
-     * <code>.Event event = 2;</code>
+     * <code>.EventEnum event = 2;</code>
      *
      * @return The event.
      */
-    org.dbsyncer.storage.binlog.proto.Event getEvent();
+    org.dbsyncer.storage.binlog.proto.EventEnum getEvent();
 
     /**
      * <code>repeated .Data data = 3;</code>

dbsyncer-storage/src/main/java/org/dbsyncer/storage/binlog/proto/BinlogMessageProto.java

@@ -43,14 +43,14 @@ public final class BinlogMessageProto {
 
     static {
         java.lang.String[] descriptorData = {
-                "\n\030BinlogMessageProto.proto\"S\n\rBinlogMess" +
-                        "age\022\026\n\016table_group_id\030\001 \001(\t\022\025\n\005event\030\002 \001" +
-                        "(\0162\006.Event\022\023\n\004data\030\003 \003(\0132\005.Data\"O\n\004Data\022" +
-                        "\033\n\003row\030\001 \003(\0132\016.Data.RowEntry\032*\n\010RowEntry" +
-                        "\022\013\n\003key\030\001 \001(\t\022\r\n\005value\030\002 \001(\014:\0028\001*+\n\005Even" +
-                        "t\022\n\n\006UPDATE\020\000\022\n\n\006INSERT\020\001\022\n\n\006DELETE\020\002B;\n" +
-                        "!org.dbsyncer.storage.binlog.protoB\022Binl" +
-                        "ogMessageProtoH\001P\001b\006proto3"
+                "\n\030BinlogMessageProto.proto\"W\n\rBinlogMess" +
+                        "age\022\026\n\016table_group_id\030\001 \001(\t\022\031\n\005event\030\002 \001" +
+                        "(\0162\n.EventEnum\022\023\n\004data\030\003 \003(\0132\005.Data\"O\n\004D" +
+                        "ata\022\033\n\003row\030\001 \003(\0132\016.Data.RowEntry\032*\n\010RowE" +
+                        "ntry\022\013\n\003key\030\001 \001(\t\022\r\n\005value\030\002 \001(\014:\0028\001*/\n\t" +
+                        "EventEnum\022\n\n\006UPDATE\020\000\022\n\n\006INSERT\020\001\022\n\n\006DEL" +
+                        "ETE\020\002B;\n!org.dbsyncer.storage.binlog.pro" +
+                        "toB\022BinlogMessageProtoH\001P\001b\006proto3"
         };
         descriptor = com.google.protobuf.Descriptors.FileDescriptor
                 .internalBuildGeneratedFileFrom(descriptorData,

dbsyncer-storage/src/main/java/org/dbsyncer/storage/binlog/proto/Event.java → dbsyncer-storage/src/main/java/org/dbsyncer/storage/binlog/proto/EventEnum.java

@@ -4,9 +4,9 @@
 package org.dbsyncer.storage.binlog.proto;
 
 /**
- * Protobuf enum {@code Event}
+ * Protobuf enum {@code EventEnum}
  */
-public enum Event
+public enum EventEnum
         implements com.google.protobuf.ProtocolMessageEnum {
   /**
    * <code>UPDATE = 0;</code>
@@ -51,7 +51,7 @@ public enum Event
    * @deprecated Use {@link #forNumber(int)} instead.
    */
   @java.lang.Deprecated
-  public static Event valueOf(int value) {
+  public static EventEnum valueOf(int value) {
     return forNumber(value);
   }
 
@@ -59,7 +59,7 @@ public enum Event
    * @param value The numeric wire value of the corresponding enum entry.
    * @return The enum associated with the given numeric wire value.
    */
-  public static Event forNumber(int value) {
+  public static EventEnum forNumber(int value) {
     switch (value) {
       case 0:
         return UPDATE;
@@ -72,16 +72,16 @@ public enum Event
     }
   }
 
-  public static com.google.protobuf.Internal.EnumLiteMap<Event>
+  public static com.google.protobuf.Internal.EnumLiteMap<EventEnum>
   internalGetValueMap() {
     return internalValueMap;
   }
 
   private static final com.google.protobuf.Internal.EnumLiteMap<
-          Event> internalValueMap =
-          new com.google.protobuf.Internal.EnumLiteMap<Event>() {
-            public Event findValueByNumber(int number) {
-              return Event.forNumber(number);
+          EventEnum> internalValueMap =
+          new com.google.protobuf.Internal.EnumLiteMap<EventEnum>() {
+            public EventEnum findValueByNumber(int number) {
+              return EventEnum.forNumber(number);
             }
           };
 
@@ -104,9 +104,9 @@ public enum Event
     return org.dbsyncer.storage.binlog.proto.BinlogMessageProto.getDescriptor().getEnumTypes().get(0);
   }
 
-  private static final Event[] VALUES = values();
+  private static final EventEnum[] VALUES = values();
 
-  public static Event valueOf(
+  public static EventEnum valueOf(
           com.google.protobuf.Descriptors.EnumValueDescriptor desc) {
     if (desc.getType() != getDescriptor()) {
       throw new java.lang.IllegalArgumentException(
@@ -120,10 +120,10 @@ public enum Event
 
   private final int value;
 
-  private Event(int value) {
+  private EventEnum(int value) {
     this.value = value;
   }
 
-  // @@protoc_insertion_point(enum_scope:Event)
+  // @@protoc_insertion_point(enum_scope:EventEnum)
 }
 

dbsyncer-storage/src/main/proto/BinlogMessageProto.proto

@@ -7,11 +7,11 @@ option optimize_for = SPEED;
 
 message BinlogMessage {
     string table_group_id = 1;
-    Event event = 2;
+    EventEnum event = 2;
     repeated Data data = 3;
 }
 
-enum Event {
+enum EventEnum {
     UPDATE = 0;
     INSERT = 1;
     DELETE = 2;

dbsyncer-storage/src/main/test/BinlogMessageTest.java

@@ -3,7 +3,7 @@ import com.google.protobuf.InvalidProtocolBufferException;
 import org.dbsyncer.common.util.DateFormatUtil;
 import org.dbsyncer.storage.binlog.proto.BinlogMessage;
 import org.dbsyncer.storage.binlog.proto.Data;
-import org.dbsyncer.storage.binlog.proto.Event;
+import org.dbsyncer.storage.binlog.proto.EventEnum;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -26,7 +26,7 @@ public class BinlogMessageTest {
 
         BinlogMessage build = BinlogMessage.newBuilder()
                 .setTableGroupId("123456788888")
-                .setEvent(Event.UPDATE)
+                .setEvent(EventEnum.UPDATE)
                 .addData(Data.newBuilder().putRow("aaa", ByteString.copyFromUtf8("hello,中国")).putRow("aaa111", ByteString.copyFromUtf8(s)))
                 .build();
 

dbsyncer-web/src/main/resources/application.properties

@@ -13,6 +13,7 @@ server.servlet.context-path=/
 #dbsyncer.storage.support.mysql.config.url=jdbc:mysql://127.0.0.1:3306/dbsyncer?rewriteBatchedStatements=true&seUnicode=true&characterEncoding=UTF8&serverTimezone=Asia/Shanghai&useSSL=false&verifyServerCertificate=false
 #dbsyncer.storage.support.mysql.config.username=root
 #dbsyncer.storage.support.mysql.config.password=123
+dbsyncer.parser.writer.buffer.actuator.enabled=true
 #dbsyncer.parser.flush.full.enabled=true
 #monitor
 management.endpoints.web.base-path=/app