service

Documentation

Index

• type Canal
  • func NewCanal(config *conf.Config) (c *Canal)
  • func (c *Canal) CheckMaster(addr, user, pwd string) (name string, pos int64, err error)
  • func (c *Canal) Close()
  • func (c *Canal) Error() string
  • func (c *Canal) Errors() (ies map[string]string)
  • func (c *Canal) PosSync(addr string) (err error)
• type HBaseInstance
  • func NewHBaseInstance(c *conf.HBaseInsConf) (ins *HBaseInstance, err error)
  • func (ins *HBaseInstance) Close()
  • func (ins *HBaseInstance) Error() string
  • func (ins *HBaseInstance) Start()
  • func (ins *HBaseInstance) String() string
• type Instance
  • func NewInstance(c *conf.InsConf) (ins *Instance, err error)
  • func (ins *Instance) Check(ev *canal.RowsEvent) (ts []*Target)
  • func (ins *Instance) Close()
  • func (ins *Instance) Error() string
  • func (ins *Instance) OnDDL(pos mysql.Position, qe *replication.QueryEvent) error
  • func (ins *Instance) OnGTID(mysql.GTIDSet) error
  • func (ins *Instance) OnPosSynced(pos mysql.Position, force bool) error
  • func (ins *Instance) OnRotate(re *replication.RotateEvent) error
  • func (ins *Instance) OnRow(ev *canal.RowsEvent) error
  • func (ins *Instance) OnXID(mysql.Position) error
  • func (ins *Instance) Start()
  • func (ins *Instance) String() string
• type Table
• type Target
  • func NewTarget(db *conf.Database) (t *Target)

Types

type Canal

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

Canal is canal.

func NewCanal

func NewCanal(config *conf.Config) (c *Canal)

NewCanal load config and start canal instance.

func (*Canal) CheckMaster

func (c *Canal) CheckMaster(addr, user, pwd string) (name string, pos int64, err error)

CheckMaster check master status.

func (*Canal) Close

func (c *Canal) Close()

Close close canal instance

func (*Canal) Error

func (c *Canal) Error() string

Error returns canal error.

func (*Canal) Errors

func (c *Canal) Errors() (ies map[string]string)

Errors returns instance errors.

func (*Canal) PosSync

func (c *Canal) PosSync(addr string) (err error)

PosSync sync newewst bin_pos.

type HBaseInstance

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

HBaseInstance hbase canal instance

func NewHBaseInstance

func NewHBaseInstance(c *conf.HBaseInsConf) (ins *HBaseInstance, err error)

NewHBaseInstance new canal instance

func (*HBaseInstance) Close

func (ins *HBaseInstance) Close()

Close close instance

func (*HBaseInstance) Error

func (ins *HBaseInstance) Error() string

Error returns instance error.

func (*HBaseInstance) Start

func (ins *HBaseInstance) Start()

Start start binlog receive

func (*HBaseInstance) String

func (ins *HBaseInstance) String() string

type Instance

type Instance struct {
	*canal.Canal
	// contains filtered or unexported fields
}

Instance canal instance

func NewInstance

func NewInstance(c *conf.InsConf) (ins *Instance, err error)

NewInstance new canal instance

func (*Instance) Check

func (ins *Instance) Check(ev *canal.RowsEvent) (ts []*Target)

Check filter row event

func (*Instance) Close

func (ins *Instance) Close()

Close close instance

func (*Instance) Error

func (ins *Instance) Error() string

Error returns instance error.

func (*Instance) OnDDL

func (ins *Instance) OnDDL(pos mysql.Position, qe *replication.QueryEvent) error

OnDDL OnDDL

func (*Instance) OnGTID

func (ins *Instance) OnGTID(mysql.GTIDSet) error

OnGTID OnGTID

func (*Instance) OnPosSynced

func (ins *Instance) OnPosSynced(pos mysql.Position, force bool) error

OnPosSynced OnPosSynced

func (*Instance) OnRotate

func (ins *Instance) OnRotate(re *replication.RotateEvent) error

OnRotate OnRotate

func (*Instance) OnRow

func (ins *Instance) OnRow(ev *canal.RowsEvent) error

OnRow send the envent to table

func (*Instance) OnXID

func (ins *Instance) OnXID(mysql.Position) error

OnXID OnXID

func (*Instance) Start

func (ins *Instance) Start()

Start start binlog receive

func (*Instance) String

func (ins *Instance) String() string

type Table

type Table struct {
	PrimaryKey []string        // kafka msg key
	OmitField  map[string]bool // field will be ignored in table
	OmitAction map[string]bool // action will be ignored in table
	// contains filtered or unexported fields
}

Table db table

type Target

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

Target databus target

func NewTarget

func NewTarget(db *conf.Database) (t *Target)

NewTarget new databus target