canal: add XID handler.

Author: siddontang

canal/handler.go

@@ -10,6 +10,7 @@ type EventHandler interface {
 OnRotate(ctx context.Context, roateEvent *replication.RotateEvent) error
 OnDDL(ctx context.Context, nextPos mysql.Position, queryEvent *replication.QueryEvent) error
 OnRow(ctx context.Context, e *RowsEvent) error
+OnXID(ctx context.Context, nextPos mysql.Position) error
 String() string
 }
 
@@ -20,8 +21,9 @@ func (h *DummyEventHandler) OnRotate(context.Context, *replication.RotateEvent)
 func (h *DummyEventHandler) OnDDL(context.Context, mysql.Position, *replication.QueryEvent) error {
 return nil
 }
-func (h *DummyEventHandler) OnRow(context.Context, *RowsEvent) error { return nil }
-func (h *DummyEventHandler) String() string { return "DummyEventHandler" }
+func (h *DummyEventHandler) OnRow(context.Context, *RowsEvent) error { return nil }
+func (h *DummyEventHandler) OnXID(context.Context, mysql.Position) error { return nil }
+func (h *DummyEventHandler) String() string { return "DummyEventHandler" }
 
 // `SetEventHandler` registers the sync handler, you must register your
 // own handler before starting Canal.

canal/sync.go

@@ -72,6 +72,9 @@ func (c *Canal) startSyncBinlog() error {
 continue
 case *replication.XIDEvent:
 // try to save the position later
+if err := c.eventHandler.OnXID(c.ctx, pos); err != nil {
+return errors.Trace(err)
+}
 case *replication.QueryEvent:
 // handle alert table query
 if mb := expAlterTable.FindSubmatch(e.Query); mb != nil {