add canal execution (go-mysql-org#98)

Author: siddontang

Makefile

@@ -4,6 +4,7 @@ build:
 rm -rf vendor && ln -s _vendor/vendor vendor
 go build -o bin/go-mysqlbinlog cmd/go-mysqlbinlog/main.go 
 go build -o bin/go-mysqldump cmd/go-mysqldump/main.go 
+go build -o bin/go-canal cmd/go-canal/main.go 
 rm -rf vendor
 
 test:

canal/canal.go

@@ -4,7 +4,6 @@ import (
 "fmt"
 "io/ioutil"
 "os"
-"os/exec"
 "path"
 "strconv"
 "strings"
@@ -97,9 +96,10 @@ func (c *Canal) prepareDumper() error {
 
 if c.dumper, err = dump.NewDumper(dumpPath,
 c.cfg.Addr, c.cfg.User, c.cfg.Password); err != nil {
-if err != exec.ErrNotFound {
-return errors.Trace(err)
-}
+return errors.Trace(err)
+}
+
+if c.dumper == nil {
 //no mysqldump, use binlog only
 return nil
 }

canal/rows.go

@@ -1,6 +1,8 @@
 package canal
 
 import (
+"fmt"
+
 "github.com/juju/errors"
 "github.com/siddontang/go-mysql/schema"
 )
@@ -50,3 +52,8 @@ func GetPKValues(table *schema.Table, row []interface{}) ([]interface{}, error)
 
 return values, nil
 }
+
+// String implements fmt.Stringer interface.
+func (r *RowsEvent) String() string {
+return fmt.Sprintf("%s %s %v", r.Action, r.Table, r.Rows)
+}

cmd/go-canal/main.go

@@ -0,0 +1,99 @@
+package main
+
+import (
+"flag"
+"fmt"
+"os"
+"os/signal"
+"strings"
+"syscall"
+
+"github.com/siddontang/go-mysql/canal"
+)
+
+var host = flag.String("host", "127.0.0.1", "MySQL host")
+var port = flag.Int("port", 3306, "MySQL port")
+var user = flag.String("user", "root", "MySQL user, must have replication privilege")
+var password = flag.String("password", "", "MySQL password")
+
+var flavor = flag.String("flavor", "mysql", "Flavor: mysql or mariadb")
+
+var dataDir = flag.String("data-dir", "./var", "Path to store data, like master.info")
+
+var serverID = flag.Int("server-id", 101, "Unique Server ID")
+var mysqldump = flag.String("mysqldump", "mysqldump", "mysqldump execution path")
+
+var dbs = flag.String("dbs", "test", "dump databases, seperated by comma")
+var tables = flag.String("tables", "", "dump tables, seperated by comma, will overwrite dbs")
+var tableDB = flag.String("table_db", "test", "database for dump tables")
+var ignoreTables = flag.String("ignore_tables", "", "ignore tables, must be database.table format, separated by comma")
+
+func main() {
+cfg := canal.NewDefaultConfig()
+cfg.Addr = fmt.Sprintf("%s:%d", *host, *port)
+cfg.User = *user
+cfg.Password = *password
+cfg.Flavor = *flavor
+cfg.DataDir = *dataDir
+
+cfg.ServerID = uint32(*serverID)
+cfg.Dump.ExecutionPath = *mysqldump
+cfg.Dump.DiscardErr = false
+
+c, err := canal.NewCanal(cfg)
+if err != nil {
+fmt.Printf("create canal err %v", err)
+os.Exit(1)
+}
+
+if len(*ignoreTables) == 0 {
+subs := strings.Split(*ignoreTables, ",")
+for _, sub := range subs {
+if seps := strings.Split(sub, "."); len(seps) == 2 {
+c.AddDumpIgnoreTables(seps[0], seps[1])
+}
+}
+}
+
+if len(*tables) > 0 && len(*tableDB) > 0 {
+subs := strings.Split(*tables, ",")
+c.AddDumpTables(*tableDB, subs...)
+} else if len(*dbs) > 0 {
+subs := strings.Split(*dbs, ",")
+c.AddDumpDatabases(subs...)
+}
+
+c.RegRowsEventHandler(&handler{})
+
+err = c.Start()
+if err != nil {
+fmt.Printf("start canal err %V", err)
+os.Exit(1)
+}
+
+sc := make(chan os.Signal, 1)
+signal.Notify(sc,
+os.Kill,
+os.Interrupt,
+syscall.SIGHUP,
+syscall.SIGINT,
+syscall.SIGTERM,
+syscall.SIGQUIT)
+
+<-sc
+
+c.Close()
+}
+
+type handler struct {
+}
+
+func (h *handler) Do(e *canal.RowsEvent) error {
+fmt.Printf("%v\n", e)
+
+return nil
+}
+
+func (h *handler) String() string {
+return "TestHandler"
+}

cmd/go-mysqldump/main.go

@@ -19,15 +19,15 @@ var output = flag.String("o", "", "dump output, empty for stdout")
 var dbs = flag.String("dbs", "", "dump databases, seperated by comma")
 var tables = flag.String("tables", "", "dump tables, seperated by comma, will overwrite dbs")
 var tableDB = flag.String("table_db", "", "database for dump tables")
-var ignoreTables = flag.String("ignore_tables", "", "ignore tables, must be database.table format, seperated by comma")
+var ignoreTables = flag.String("ignore_tables", "", "ignore tables, must be database.table format, separated by comma")
 
 func main() {
 flag.Parse()
 
 d, err := dump.NewDumper(*execution, *addr, *user, *password)
 if err != nil {
 fmt.Printf("Create Dumper error %v\n", errors.ErrorStack(err))
-return
+os.Exit(1)
 }
 
 if len(*ignoreTables) == 0 {
@@ -53,14 +53,14 @@ func main() {
 f, err = os.OpenFile(*output, os.O_CREATE|os.O_WRONLY, 0644)
 if err != nil {
 fmt.Printf("Open file error %v\n", errors.ErrorStack(err))
-return
+os.Exit(1)
 }
 }
 
 defer f.Close()
 
 if err = d.Dump(f); err != nil {
 fmt.Printf("Dump MySQL error %v\n", errors.ErrorStack(err))
-return
+os.Exit(1)
 }
 }

dump/dump.go

@@ -32,7 +32,7 @@ type Dumper struct {
 
 func NewDumper(executionPath string, addr string, user string, password string) (*Dumper, error) {
 if len(executionPath) == 0 {
-executionPath = "mysqldump"
+return nil, nil
 }
 
 path, err := exec.LookPath(executionPath)

dump/dump_test.go

@@ -36,6 +36,7 @@ func (s *schemaTestSuite) SetUpSuite(c *C) {
 
 s.d, err = NewDumper(*execution, fmt.Sprintf("%s:%d", *host, *port), "root", "")
 c.Assert(err, IsNil)
+c.Assert(s.d, NotNil)
 
 s.d.SetErrOut(os.Stderr)