mysql

Documentation

Index

• Variables
• func GetInt(row map[string]interface{}, key string) int
• func GetString(row map[string]interface{}, key string) string
• func GetTime(row map[string]interface{}, key string) time.Time
• type BinlogHandler
• type Database
• type DispatchHandler
  • func NewDispatchHandler(dbs []*Database) *DispatchHandler
  • func (m *DispatchHandler) GetGTidSet() mysql.GTIDSet
  • func (m *DispatchHandler) OnPosSynced(eh *replication.EventHeader, p mysql.Position, g mysql.GTIDSet, force bool) error
  • func (m *DispatchHandler) OnRow(e *canal.RowsEvent) error
  • func (m *DispatchHandler) SetGTidSet(g mysql.GTIDSet)
• type DummyBinLogHandler
  • func (h *DummyBinLogHandler) GetActions() []string
  • func (h *DummyBinLogHandler) GetDbName() string
  • func (h *DummyBinLogHandler) GetTableName() string
  • func (h *DummyBinLogHandler) OnRow(*RowEventData) error
• type OnRowHandler
• type Options
  • func NewDefaultOpt() *Options
• type RowEventData
• type RowEventMonitor
  • func NewRowEventMonitor(opt *Options) (*RowEventMonitor, error)
  • func (m *RowEventMonitor) AddHandler(handlers ...OnRowHandler) error
  • func (m *RowEventMonitor) SetStore(store cdcstore.MysqlPosInterface)
  • func (m *RowEventMonitor) StartMonitor() error
• type TableInfo

Variables

var KEmptyTime = time.Date(1970, 1, 1, 8, 0, 0, 0, time.Local)

Functions

func GetInt

func GetInt(row map[string]interface{}, key string) int

func GetString

func GetString(row map[string]interface{}, key string) string

func GetTime

func GetTime(row map[string]interface{}, key string) time.Time

Types

type BinlogHandler

type BinlogHandler interface {
	GetDbName() string
	GetTableName() string
	GetActions() []string
}

type Database

type Database struct {
	Name   string       `yaml:"name"`
	Tables []*TableInfo `yaml:"tables"`
}

type DispatchHandler

type DispatchHandler struct {
	cdcstore.MysqlPosInterface
	canal.DummyEventHandler
	// contains filtered or unexported fields
}

func NewDispatchHandler

func NewDispatchHandler(dbs []*Database) *DispatchHandler

func (*DispatchHandler) GetGTidSet

func (m *DispatchHandler) GetGTidSet() mysql.GTIDSet

func (*DispatchHandler) OnPosSynced

func (m *DispatchHandler) OnPosSynced(eh *replication.EventHeader, p mysql.Position, g mysql.GTIDSet, force bool) error

func (*DispatchHandler) OnRow

func (m *DispatchHandler) OnRow(e *canal.RowsEvent) error

func (*DispatchHandler) SetGTidSet

func (m *DispatchHandler) SetGTidSet(g mysql.GTIDSet)

type DummyBinLogHandler

type DummyBinLogHandler struct {
	DbName    string
	TableName string
	Actions   []string
}

func (*DummyBinLogHandler) GetActions

func (h *DummyBinLogHandler) GetActions() []string

func (*DummyBinLogHandler) GetDbName

func (h *DummyBinLogHandler) GetDbName() string

func (*DummyBinLogHandler) GetTableName

func (h *DummyBinLogHandler) GetTableName() string

func (*DummyBinLogHandler) OnRow

func (h *DummyBinLogHandler) OnRow(*RowEventData) error

type OnRowHandler

type OnRowHandler interface {
	BinlogHandler
	OnRow(*RowEventData) error
}

type Options

type Options struct {
	Addr       string      `yaml:"addr"`
	User       string      `yaml:"user"`
	Pwd        string      `yaml:"pwd"`
	Charset    string      `yaml:"charset"`
	MaxIdleNum int         `yaml:"max_idle_num"`
	MaxConnNum int         `yaml:"max_conn_num"`
	DefaultDb  string      `yaml:"default_db"`
	Databases  []*Database `yaml:"databases"`
}

func NewDefaultOpt

func NewDefaultOpt() *Options

type RowEventData

type RowEventData struct {
	Database  string
	Table     string
	Action    string
	Timestamp uint32
	Pos       uint32
	Old       map[string]interface{}
	Row       map[string]interface{}
	Err       error
}

type RowEventMonitor

type RowEventMonitor struct {
	// contains filtered or unexported fields
}

func NewRowEventMonitor

func NewRowEventMonitor(opt *Options) (*RowEventMonitor, error)

func (*RowEventMonitor) AddHandler

func (m *RowEventMonitor) AddHandler(handlers ...OnRowHandler) error

func (*RowEventMonitor) SetStore

func (m *RowEventMonitor) SetStore(store cdcstore.MysqlPosInterface)

func (*RowEventMonitor) StartMonitor

func (m *RowEventMonitor) StartMonitor() error

type TableInfo

type TableInfo struct {
	Name          string   `yaml:"name"`
	Actions       []string `yaml:"actions"`
	ColumnMapping map[string]*columnInfo
}