Merge pull request #3 from instructure/MCE-18274_port_fs_loader

MCE-18274 port fs loader

Author: smashleon23

canal/canal.go

@@ -6,6 +6,7 @@ import (
 "io"
 "net"
 "os"
+"path"
 "regexp"
 "strconv"
 "strings"
@@ -83,6 +84,12 @@ func NewCanal(cfg *Config) (*Canal, error) {
 }
 c.master = &masterInfo{logger: c.cfg.Logger}
 
+// allow the master info to be loaded from a configured loader
+c.master.infoLoader = cfg.InfoLoader
+if c.master.infoLoader == nil {
+c.master.infoLoader = NewFsInfoLoader(c.masterInfoPath())
+}
+
 c.delay = new(uint32)
 
 var err error
@@ -506,6 +513,10 @@ func (c *Canal) connect(options ...func(*client.Conn)) (*client.Conn, error) {
 c.cfg.User, c.cfg.Password, "", c.cfg.Dialer, options...)
 }
 
+func (c *Canal) masterInfoPath() string {
+return path.Join(c.cfg.DataDir, "master.info")
+}
+
 // Execute a SQL
 func (c *Canal) Execute(cmd string, args ...interface{}) (rr *mysql.Result, err error) {
 c.connLock.Lock()

canal/config.go

@@ -57,6 +57,7 @@ type Config struct {
 Charset string `toml:"charset"`
 ServerID uint32 `toml:"server_id"`
 Flavor string `toml:"flavor"`
+DataDir string `toml:"data_dir"`
 HeartbeatPeriod time.Duration `toml:"heartbeat_period"`
 ReadTimeout time.Duration `toml:"read_timeout"`
 
@@ -99,6 +100,9 @@ type Config struct {
 
 //Set Localhost
 Localhost string
+
+// allow the master info loading to be configured
+InfoLoader MasterInfoLoader
 }
 
 func NewConfigWithFile(name string) (*Config, error) {

canal/fsloader.go

@@ -0,0 +1,62 @@
+package canal
+
+import (
+"bytes"
+"os"
+
+"github.com/BurntSushi/toml"
+"github.com/instructure/mc-go-mysql/mysql"
+"github.com/pingcap/errors"
+"github.com/siddontang/go-log/log"
+"github.com/siddontang/go/ioutil2"
+)
+
+type fsInfoLoader struct {
+path string
+}
+
+func NewFsInfoLoader(path string) MasterInfoLoader {
+return &fsInfoLoader{path: path}
+}
+
+func (l *fsInfoLoader) Load(setValues MasterInfoSetter) error {
+f, err := os.Open(l.path)
+if err != nil && !os.IsNotExist(errors.Cause(err)) {
+return errors.Trace(err)
+} else if os.IsNotExist(errors.Cause(err)) {
+return nil
+}
+defer f.Close()
+
+var m masterInfo
+_, err = toml.DecodeReader(f, &m)
+
+if err != nil {
+return err
+}
+
+return setValues(m.Addr, m.pos.Name, m.pos.Pos)
+}
+
+func (l *fsInfoLoader) Save(addr, name string, position uint32, force bool) error {
+var buf bytes.Buffer
+e := toml.NewEncoder(&buf)
+pos := mysql.Position{Name: name, Pos: position}
+
+m := &masterInfo{
+Addr: addr,
+pos: pos,
+}
+
+enc_err := e.Encode(m)
+if enc_err != nil {
+log.Errorf("canal save master info to file %s err %v", l.path, enc_err)
+}
+
+var err error
+if err = ioutil2.WriteFileAtomic(l.path, buf.Bytes(), 0644); err != nil {
+log.Errorf("canal save master info to file %s err %v", l.path, err)
+}
+
+return errors.Trace(err)
+}

canal/master.go

@@ -10,13 +10,30 @@ import (
 type masterInfo struct {
 sync.RWMutex
 
+Addr string
+
 pos mysql.Position
 
 gset mysql.GTIDSet
 
 timestamp uint32
 
 logger loggers.Advanced
+
+infoLoader MasterInfoLoader
+}
+
+// abstract the way in which the master info is loaded and saved
+type MasterInfoSetter func(addr, name string, position uint32) error
+type MasterInfoLoader interface {
+Load(setValues MasterInfoSetter) error
+Save(addr, name string, position uint32, force bool) error
+}
+
+func (m *masterInfo) Setter(addr, name string, position uint32) error {
+m.Addr = addr
+m.pos = mysql.Position{Name: name, Pos: position}
+return nil
 }
 
 func (m *masterInfo) Update(pos mysql.Position) {