func (b *BinlogSyncer) StartSync(pos Position) (*BinlogStreamer, error) {
	Log.Infof("begin to sync binlog from position %s", pos)
	b.m.Lock()
	defer b.m.Unlock()
	if b.running {
		return nil, errors.Trace(errSyncRunning)
	}
	if err := b.prepareSyncPos(pos); err != nil {
		return nil, errors.Trace(err)
	}
	return b.startDumpStream(), nil
}
StartSync starts syncing from the `pos` position.
package replication
import (
	"context"
	"crypto/tls"
	"encoding/binary"
	"fmt"
	"net"
	"os"
	"sync"
	"time"
	. "github.com/flike/kingbus/log"
	"github.com/juju/errors"
	"github.com/satori/go.uuid"
	"github.com/siddontang/go-mysql/client"
	. "github.com/siddontang/go-mysql/mysql"
)
var (
	errSyncRunning = errors.New("Sync is running, must Close first")
)
type BinlogSyncerConfig struct {
	ServerID uint32
	ServerUuid string
	Flavor string
	Host string
	Port uint16
	User string
	Password string
	Localhost string
	Charset string
	SemiSyncEnabled bool
	RawModeEnabled bool
	TLSConfig *tls.Config
	ParseTime bool
	TimestampStringLocation *time.Location
	UseDecimal bool
	RecvBufferSize int
	HeartbeatPeriod time.Duration
	ReadTimeout time.Duration
	MaxReconnectAttempts int
	VerifyChecksum bool
}
type BinlogSyncer struct {
	m sync.RWMutex
	cfg BinlogSyncerConfig
	c *client.Conn
	wg sync.WaitGroup
	parser *BinlogParser
	nextPos Position
	gset GTIDSet
	running bool
	ctx    context.Context
	cancel context.CancelFunc
	lastConnectionID uint32
	retryCount int
}
func NewBinlogSyncer(cfg BinlogSyncerConfig) *BinlogSyncer {
	if cfg.ServerID == 0 {
		Log.Fatal("can't use 0 as the server ID")
	}
	pass := cfg.Password
	cfg.Password = ""
	Log.Infof("create BinlogSyncer with config %v", cfg)
	cfg.Password = pass
	b := new(BinlogSyncer)
	b.cfg = cfg
	b.parser = NewBinlogParser()
	b.parser.SetRawMode(b.cfg.RawModeEnabled)
	b.parser.SetParseTime(b.cfg.ParseTime)
	b.parser.SetTimestampStringLocation(b.cfg.TimestampStringLocation)
	b.parser.SetUseDecimal(b.cfg.UseDecimal)
	b.parser.SetVerifyChecksum(b.cfg.VerifyChecksum)
	b.running = false
	b.ctx, b.cancel = context.WithCancel(context.Background())
	return b
}
func (b *BinlogSyncer) Close() {
	b.m.Lock()
	defer b.m.Unlock()
	b.close()
}
func (b *BinlogSyncer) close() {
	if b.isClosed() {
		return
	}
	Log.Info("syncer is closing...")
	b.running = false
	b.cancel()
	if b.c != nil {
		b.c.SetReadDeadline(time.Now().Add(100 * time.Millisecond))
	}
	b.wg.Wait()
	if b.c != nil {
		b.c.Close()
	}
	Log.Info("syncer is closed")
}
func (b *BinlogSyncer) isClosed() bool {
	select {
	case <-b.ctx.Done():
		return true
	default:
		return false
	}
}
func (b *BinlogSyncer) registerSlave() error {
	if b.c != nil {
		b.c.Close()
	}
	Log.Infof("register slave for master server %s:%d", b.cfg.Host, b.cfg.Port)
	var err error
	b.c, err = client.Connect(fmt.Sprintf("%s:%d", b.cfg.Host, b.cfg.Port), b.cfg.User, b.cfg.Password, "", func(c *client.Conn) {
		c.TLSConfig = b.cfg.TLSConfig
	})
	if err != nil {
		return errors.Trace(err)
	}
	if len(b.cfg.Charset) != 0 {
		b.c.SetCharset(b.cfg.Charset)
	}
	if b.cfg.ReadTimeout > 0 {
		b.c.SetReadDeadline(time.Now().Add(b.cfg.ReadTimeout))
	}
	if b.cfg.RecvBufferSize > 0 {
		if tcp, ok := b.c.Conn.Conn.(*net.TCPConn); ok {
			tcp.SetReadBuffer(b.cfg.RecvBufferSize)
		}
	}
	if b.lastConnectionID > 0 {
		cmd := fmt.Sprintf("KILL %d", b.lastConnectionID)
		if _, err := b.c.Execute(cmd); err != nil {
			Log.Errorf("kill connection %d error %v", b.lastConnectionID, err)
			if code := ErrorCode(err.Error()); code != ER_NO_SUCH_THREAD {
				return errors.Trace(err)
			}
		}
		Log.Infof("kill last connection id %d", b.lastConnectionID)
	}
	b.lastConnectionID = b.c.GetConnectionID()
	if r, err := b.c.Execute("SELECT @@global.binlog_checksum"); err != nil {
		return errors.Trace(err)
	} else {
		s, _ := r.GetString(0, 0)
		Log.Debugf("global.binlog_checksum:%s", s)
		if s == "CRC32" {
			if _, err = b.c.Execute(`SET @master_binlog_checksum='CRC32'`); err != nil {
				return errors.Trace(err)
			}
			if err = b.c.SetSlaveUuid(b.cfg.ServerUuid); err != nil {
				return errors.Trace(err)
			}
		} else {
			Log.Errorf("master global.binlog_checksum is not CRC32")
			return fmt.Errorf("master binlog_checksum is not CRC32")
		}
	}
	if b.cfg.Flavor == MariaDBFlavor {
		if _, err := b.c.Execute("SET @mariadb_slave_capability=4"); err != nil {
			return errors.Errorf("failed to set @mariadb_slave_capability=4: %v", err)
		}
	}
	if b.cfg.HeartbeatPeriod > 0 {
		_, err = b.c.Execute(fmt.Sprintf("SET @master_heartbeat_period=%d;", b.cfg.HeartbeatPeriod))
		if err != nil {
			Log.Error("failed to set @master_heartbeat_period=%d", b.cfg.HeartbeatPeriod, err)
			return errors.Trace(err)
		}
	}
	if err = b.writeRegisterSlaveCommand(); err != nil {
		return errors.Trace(err)
	}
	if _, err = b.c.ReadOKPacket(); err != nil {
		return errors.Trace(err)
	}
	return nil
}
func (b *BinlogSyncer) enableSemiSync() error {
	if !b.cfg.SemiSyncEnabled {
		return nil
	}
	if r, err := b.c.Execute("SHOW VARIABLES LIKE 'rpl_semi_sync_master_enabled';"); err != nil {
		return errors.Trace(err)
	} else {
		s, _ := r.GetString(0, 1)
		if s != "ON" {
			Log.Errorf("master does not support semi synchronous replication, use no semi-sync")
			b.cfg.SemiSyncEnabled = false
			return nil
		}
	}
	_, err := b.c.Execute(`SET @rpl_semi_sync_slave = 1;`)
	if err != nil {
		return errors.Trace(err)
	}
	return nil
}
func (b *BinlogSyncer) prepare() error {
	if b.isClosed() {
		return errors.Trace(ErrSyncClosed)
	}
	if err := b.registerSlave(); err != nil {
		return errors.Trace(err)
	}
	if err := b.enableSemiSync(); err != nil {
		return errors.Trace(err)
	}
	return nil
}
func (b *BinlogSyncer) startDumpStream() *BinlogStreamer {
	b.running = true
	s := newBinlogStreamer()
	b.wg.Add(1)
	go b.onStream(s)
	return s
}
func (b *BinlogSyncer) GetNextPosition() Position {
	return b.nextPos
}