Merge pull request #7 from axfor/feature-axx

added query event in dummyEventHandler1

Author: axfor

canal/handler.go

@@ -17,6 +17,8 @@ type EventHandler interface {
 OnGTID(header *replication.EventHeader, gtid mysql.GTIDSet) error
 // OnPosSynced Use your own way to sync position. When force is true, sync position immediately.
 OnPosSynced(header *replication.EventHeader, pos mysql.Position, set mysql.GTIDSet, force bool) error
+// OnQueryEvent is query event include(create user,drop user,create index event,etd.)
+OnQueryEvent(header *replication.EventHeader, queryEvent *replication.QueryEvent) error
 String() string
 }
 
@@ -38,7 +40,9 @@ func (h *DummyEventHandler) OnGTID(*replication.EventHeader, mysql.GTIDSet) erro
 func (h *DummyEventHandler) OnPosSynced(*replication.EventHeader, mysql.Position, mysql.GTIDSet, bool) error {
 return nil
 }
-
+func (h *DummyEventHandler) OnQueryEvent(header *replication.EventHeader, queryEvent *replication.QueryEvent) error {
+return nil
+}
 func (h *DummyEventHandler) String() string { return "DummyEventHandler" }
 
 // `SetEventHandler` registers the sync handler, you must register your

canal/sync.go

@@ -142,22 +142,9 @@ func (c *Canal) runSyncBinlog() error {
 continue
 }
 for _, stmt := range stmts {
-nodes := parseStmt(stmt)
-for _, node := range nodes {
-if node.db == "" {
-node.db = string(e.Schema)
-}
-if err = c.updateTable(ev.Header, node.db, node.table); err != nil {
-return errors.Trace(err)
-}
-}
-if len(nodes) > 0 {
-savePos = true
-force = true
-// Now we only handle Table Changed DDL, maybe we will support more later.
-if err = c.eventHandler.OnDDL(ev.Header, pos, e); err != nil {
-return errors.Trace(err)
-}
+err := c.handleQueryEvent(ev, e, stmt, pos, &savePos, &force)
+if err != nil {
+c.cfg.Logger.Errorf("handle query event(%s) err %v", e.Query, err)
 }
 }
 if savePos && e.GSet != nil {
@@ -336,3 +323,45 @@ func (c *Canal) CatchMasterPos(timeout time.Duration) error {
 
 return c.WaitUntilPos(pos, timeout)
 }
+
+// handleQueryEvent is handle some common query events (e.g., DDL,CREATE or DROP USER,GRANT),
+// others use UnknownQueryEvent unified callbacks to expose to users
+func (c *Canal) handleQueryEvent(ev *replication.BinlogEvent, e *replication.QueryEvent,
+stmt ast.StmtNode, pos mysql.Position, savePos, force *bool) error {
+switch t := stmt.(type) {
+case *ast.RenameTableStmt, *ast.AlterTableStmt, *ast.DropTableStmt, *ast.CreateTableStmt, *ast.TruncateTableStmt:
+return c.handleDDLEvent(ev, e, t, pos, savePos, force)
+default:
+return c.handleUnknownQueryEvent(ev, e, t, pos, savePos, force)
+}
+}
+
+func (c *Canal) handleDDLEvent(ev *replication.BinlogEvent, e *replication.QueryEvent,
+stmt ast.StmtNode, pos mysql.Position, savePos, force *bool) error {
+nodes := parseStmt(stmt)
+for _, node := range nodes {
+if node.db == "" {
+node.db = string(e.Schema)
+}
+if err := c.updateTable(ev.Header, node.db, node.table); err != nil {
+return errors.Trace(err)
+}
+}
+if len(nodes) > 0 {
+*savePos = true
+*force = true
+// Now we only handle Table Changed DDL, maybe we will support more later.
+if err := c.eventHandler.OnDDL(ev.Header, pos, e); err != nil {
+return errors.Trace(err)
+}
+}
+return nil
+}
+
+func (c *Canal) handleUnknownQueryEvent(ev *replication.BinlogEvent, e *replication.QueryEvent,
+stmt ast.StmtNode, pos mysql.Position, savePos, force *bool) error {
+if err := c.eventHandler.OnQueryEvent(ev.Header, e); err != nil {
+return errors.Trace(err)
+}
+return nil
+}

client/resp.go

@@ -336,15 +336,15 @@ func (c *Conn) readResultsetStreaming(data []byte, binary bool, result *Result,
 return nil
 }
 
-func (c *Conn) readResultColumns(result *Result) (err error) {
+func (c *Conn) readResultColumns(result *Result) error {
 var i = 0
 var data []byte
-
+var err error
 for {
 rawPkgLen := len(result.RawPkg)
 result.RawPkg, err = c.ReadPacketReuseMem(result.RawPkg)
 if err != nil {
-return
+return err
 }
 data = result.RawPkg[rawPkgLen:]
 
@@ -361,15 +361,15 @@ func (c *Conn) readResultColumns(result *Result) (err error) {
 err = ErrMalformPacket
 }
 
-return
+return err
 }
 
 if result.Fields[i] == nil {
 result.Fields[i] = &Field{}
 }
 err = result.Fields[i].Parse(data)
 if err != nil {
-return
+return err
 }
 
 result.FieldNames[hack.String(result.Fields[i].Name)] = i
@@ -378,14 +378,15 @@ func (c *Conn) readResultColumns(result *Result) (err error) {
 }
 }
 
-func (c *Conn) readResultRows(result *Result, isBinary bool) (err error) {
+func (c *Conn) readResultRows(result *Result, isBinary bool) error {
 var data []byte
+var err error
 
 for {
 rawPkgLen := len(result.RawPkg)
 result.RawPkg, err = c.ReadPacketReuseMem(result.RawPkg)
 if err != nil {
-return
+return err
 }
 data = result.RawPkg[rawPkgLen:]
 
@@ -425,16 +426,17 @@ func (c *Conn) readResultRows(result *Result, isBinary bool) (err error) {
 return nil
 }
 
-func (c *Conn) readResultRowsStreaming(result *Result, isBinary bool, perRowCb SelectPerRowCallback) (err error) {
+func (c *Conn) readResultRowsStreaming(result *Result, isBinary bool, perRowCb SelectPerRowCallback) error {
 var (
 data []byte
 row []FieldValue
+err error
 )
 
 for {
 data, err = c.ReadPacketReuseMem(data[:0])
 if err != nil {
-return
+return err
 }
 
 // EOF Packet

mysql/field.go

@@ -41,50 +41,51 @@ const (
 FieldValueTypeString
 )
 
-func (f *Field) Parse(p FieldData) (err error) {
+func (f *Field) Parse(p FieldData) error {
 f.Data = p
 
 var n int
+var err error
 pos := 0
 //skip catelog, always def
 n, err = SkipLengthEncodedString(p)
 if err != nil {
-return
+return err
 }
 pos += n
 
 //schema
 f.Schema, _, n, err = LengthEncodedString(p[pos:])
 if err != nil {
-return
+return err
 }
 pos += n
 
 //table
 f.Table, _, n, err = LengthEncodedString(p[pos:])
 if err != nil {
-return
+return err
 }
 pos += n
 
 //org_table
 f.OrgTable, _, n, err = LengthEncodedString(p[pos:])
 if err != nil {
-return
+return err
 }
 pos += n
 
 //name
 f.Name, _, n, err = LengthEncodedString(p[pos:])
 if err != nil {
-return
+return err
 }
 pos += n
 
 //org_name
 f.OrgName, _, n, err = LengthEncodedString(p[pos:])
 if err != nil {
-return
+return err
 }
 pos += n
 
@@ -123,7 +124,7 @@ func (f *Field) Parse(p FieldData) (err error) {
 
 if pos+int(f.DefaultValueLength) > len(p) {
 err = ErrMalformPacket
-return
+return err
 }
 
 //default value string[$len]