canal: remove no need context now

Author: siddontang

canal/canal_test.go

@@ -10,7 +10,6 @@ import (
 "github.com/ngaut/log"
 . "github.com/pingcap/check"
 "github.com/siddontang/go-mysql/mysql"
-"golang.org/x/net/context"
 )
 
 var testHost = flag.String("host", "127.0.0.1", "MySQL host")
@@ -74,7 +73,7 @@ type testEventHandler struct {
 DummyEventHandler
 }
 
-func (h *testEventHandler) Do(_ context.Context, e *RowsEvent) error {
+func (h *testEventHandler) Do(e *RowsEvent) error {
 log.Infof("%s %v\n", e.Action, e.Rows)
 return nil
 }

canal/dump.go

@@ -64,7 +64,7 @@ func (h *dumpParseHandler) Data(db string, table string, values []string) error
 }
 
 events := newRowsEvent(tableInfo, InsertAction, [][]interface{}{vs})
-return h.c.eventHandler.OnRow(h.c.ctx, events)
+return h.c.eventHandler.OnRow(events)
 }
 
 func (c *Canal) AddDumpDatabases(dbs ...string) {

canal/handler.go

@@ -3,27 +3,26 @@ package canal
 import (
 "github.com/siddontang/go-mysql/mysql"
 "github.com/siddontang/go-mysql/replication"
-"golang.org/x/net/context"
 )
 
 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
+OnRotate(roateEvent *replication.RotateEvent) error
+OnDDL(nextPos mysql.Position, queryEvent *replication.QueryEvent) error
+OnRow(e *RowsEvent) error
+OnXID(nextPos mysql.Position) error
 String() string
 }
 
 type DummyEventHandler struct {
 }
 
-func (h *DummyEventHandler) OnRotate(context.Context, *replication.RotateEvent) error { return nil }
-func (h *DummyEventHandler) OnDDL(context.Context, mysql.Position, *replication.QueryEvent) error {
+func (h *DummyEventHandler) OnRotate(*replication.RotateEvent) error { return nil }
+func (h *DummyEventHandler) OnDDL(mysql.Position, *replication.QueryEvent) error {
 return nil
 }
-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" }
+func (h *DummyEventHandler) OnRow(*RowsEvent) error { return nil }
+func (h *DummyEventHandler) OnXID(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

@@ -4,8 +4,6 @@ import (
 "regexp"
 "time"
 
-"golang.org/x/net/context"
-
 "github.com/juju/errors"
 "github.com/ngaut/log"
 "github.com/siddontang/go-mysql/mysql"
@@ -27,23 +25,13 @@ func (c *Canal) startSyncBinlog() error {
 return errors.Errorf("start sync replication at %v error %v", pos, err)
 }
 
-timeout := time.Second
 for {
-ctx, cancel := context.WithTimeout(c.ctx, 2*time.Second)
-ev, err := s.GetEvent(ctx)
-cancel()
-
-if err == context.DeadlineExceeded {
-timeout = 2 * timeout
-continue
-}
+ev, err := s.GetEvent(c.ctx)
 
 if err != nil {
 return errors.Trace(err)
 }
 
-timeout = time.Second
-
 curPos := pos.Pos
 //next binlog pos
 pos.Pos = ev.Header.LogPos
@@ -58,7 +46,7 @@ func (c *Canal) startSyncBinlog() error {
 pos.Pos = uint32(e.Position)
 log.Infof("rotate binlog to %s", pos)
 
-if err = c.eventHandler.OnRotate(c.ctx, e); err != nil {
+if err = c.eventHandler.OnRotate(e); err != nil {
 return errors.Trace(err)
 }
 case *replication.RowsEvent:
@@ -72,7 +60,7 @@ 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 {
+if err := c.eventHandler.OnXID(pos); err != nil {
 return errors.Trace(err)
 }
 case *replication.QueryEvent:
@@ -83,7 +71,7 @@ func (c *Canal) startSyncBinlog() error {
 }
 c.ClearTableCache(mb[1], mb[2])
 log.Infof("table structure changed, clear table cache: %s.%s\n", mb[1], mb[2])
-if err = c.eventHandler.OnDDL(c.ctx, pos, e); err != nil {
+if err = c.eventHandler.OnDDL(pos, e); err != nil {
 return errors.Trace(err)
 }
 } else {
@@ -123,7 +111,7 @@ func (c *Canal) handleRowsEvent(e *replication.BinlogEvent) error {
 return errors.Errorf("%s not supported now", e.Header.EventType)
 }
 events := newRowsEvent(t, action, ev.Rows)
-return c.eventHandler.OnRow(c.ctx, events)
+return c.eventHandler.OnRow(events)
 }
 
 func (c *Canal) WaitUntilPos(pos mysql.Position, timeout time.Duration) error {

cmd/go-canal/main.go

@@ -10,7 +10,6 @@ import (
 
 "github.com/siddontang/go-mysql/canal"
 "github.com/siddontang/go-mysql/mysql"
-"golang.org/x/net/context"
 )
 
 var host = flag.String("host", "127.0.0.1", "MySQL host")
@@ -98,7 +97,7 @@ type handler struct {
 canal.DummyEventHandler
 }
 
-func (h *handler) OnRow(_ context.Context, e *canal.RowsEvent) error {
+func (h *handler) OnRow(e *canal.RowsEvent) error {
 fmt.Printf("%v\n", e)
 
 return nil