mysql_cdc

Documentation

Index

• Constants
• type Config
• type Endpoint
• type EventHandler
  • func (h *EventHandler) OnRow(e *canal.RowsEvent) error
  • func (h *EventHandler) String() string
• type MySqlCDC
  • func (x *MySqlCDC) AddRouter(router endpointApi.Router, params ...interface{}) (string, error)
  • func (x *MySqlCDC) Close() error
  • func (x *MySqlCDC) Destroy()
  • func (x *MySqlCDC) GetDefaultConfig(newConfig Config) *canal.Config
  • func (x *MySqlCDC) Id() string
  • func (x *MySqlCDC) Init(ruleConfig types.Config, configuration types.Configuration) error
  • func (x *MySqlCDC) New() types.Node
  • func (x *MySqlCDC) Printf(format string, v ...interface{})
  • func (x *MySqlCDC) RemoveRouter(routerId string, params ...interface{}) error
  • func (x *MySqlCDC) Start() error
  • func (x *MySqlCDC) Type() string
• type RegexpRouter
• type RequestMessage
  • func (r *RequestMessage) Body() []byte
  • func (r *RequestMessage) ColumnNames() []string
  • func (r *RequestMessage) From() string
  • func (r *RequestMessage) GetError() error
  • func (r *RequestMessage) GetMsg() *types.RuleMsg
  • func (r *RequestMessage) GetParam(key string) string
  • func (r *RequestMessage) Headers() textproto.MIMEHeader
  • func (r *RequestMessage) PKColumns() []string
  • func (r *RequestMessage) SetBody(body []byte)
  • func (r *RequestMessage) SetError(err error)
  • func (r *RequestMessage) SetMsg(msg *types.RuleMsg)
  • func (r *RequestMessage) SetStatusCode(statusCode int)
• type ResponseMessage
  • func (r *ResponseMessage) Body() []byte
  • func (r *ResponseMessage) From() string
  • func (r *ResponseMessage) GetError() error
  • func (r *ResponseMessage) GetMsg() *types.RuleMsg
  • func (r *ResponseMessage) GetParam(key string) string
  • func (r *ResponseMessage) Headers() textproto.MIMEHeader
  • func (r *ResponseMessage) SetBody(body []byte)
  • func (r *ResponseMessage) SetError(err error)
  • func (r *ResponseMessage) SetMsg(msg *types.RuleMsg)
  • func (r *ResponseMessage) SetStatusCode(statusCode int)

Constants

const (
	KeyTableSchema   = "tableSchema"
	KeyTableName     = "tableName"
	KeyTableFullName = "tableFullName"
	KeyAction        = "action"
	KeyColumnNames   = "columnNames"
	KeyPkColumnNames = "pkColumnNames"
	KeyLogPos        = "logPos"
	// MatchAll 匹配所有数据
	MatchAll = "*"
)

const Type = types.EndpointTypePrefix + "mysql_cdc"

Type 组件类型

Types

type Config

type Config struct {
	// mysql服务器地址
	Server string
	//用户名
	User string
	// 密码
	Password string
	//FromOldest 是否从最旧binlog同步，否则从最新的binlog和位置同步
	FromOldest bool
	// 数据库
	Dbs []string
	// IncludeTables or ExcludeTables should contain database name.
	// IncludeTables defines the tables that will be included, if empty, all tables will be included.
	// ExcludeTables defines the tables that will be excluded from the ones defined by IncludeTables.
	// Only a table which matches IncludeTables and dismatches ExcludeTables will be processed
	// eg, IncludeTables : [".*\\.canal","test.*"], ExcludeTables : ["mysql\\..*"]
	//     this will include all database's 'canal' table, except database 'mysql'.
	// Default IncludeTables and ExcludeTables are empty, this will include all tables
	IncludeTables []string
	ExcludeTables []string

	// mysqldump execution path, like mysqldump or /usr/bin/mysqldump, etc...
	// If not set, ignore using mysqldump.
	ExecutionPath string
	//字符集
	Charset string
	//mysql or mariadb
	Flavor string
	//心跳单位秒
	Heartbeat int
	// 读超时单位秒
	ReadTimeout int
}

type Endpoint

type Endpoint = MySqlCDC

Endpoint 别名

type EventHandler

type EventHandler struct {
	canal.DummyEventHandler
	// contains filtered or unexported fields
}

func (*EventHandler) OnRow

func (h *EventHandler) OnRow(e *canal.RowsEvent) error

func (*EventHandler) String

func (h *EventHandler) String() string

type MySqlCDC

type MySqlCDC struct {
	impl.BaseEndpoint
	RuleConfig types.Config
	//Config 配置
	Config Config
	// contains filtered or unexported fields
}

MySqlCDC 接收端端点

func (*MySqlCDC) AddRouter

func (x *MySqlCDC) AddRouter(router endpointApi.Router, params ...interface{}) (string, error)

func (*MySqlCDC) Close

func (x *MySqlCDC) Close() error

func (*MySqlCDC) Destroy

func (x *MySqlCDC) Destroy()

Destroy 销毁

func (*MySqlCDC) GetDefaultConfig

func (x *MySqlCDC) GetDefaultConfig(newConfig Config) *canal.Config

func (*MySqlCDC) Id

func (x *MySqlCDC) Id() string

func (*MySqlCDC) Init

func (x *MySqlCDC) Init(ruleConfig types.Config, configuration types.Configuration) error

Init 初始化

func (*MySqlCDC) New

func (x *MySqlCDC) New() types.Node

func (*MySqlCDC) Printf

func (x *MySqlCDC) Printf(format string, v ...interface{})

func (*MySqlCDC) RemoveRouter

func (x *MySqlCDC) RemoveRouter(routerId string, params ...interface{}) error

func (*MySqlCDC) Start

func (x *MySqlCDC) Start() error

func (*MySqlCDC) Type

func (x *MySqlCDC) Type() string

Type 组件类型

type RegexpRouter

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

RegexpRouter 正则表达式路由

type RequestMessage

type RequestMessage struct {
	Table *schema.Table
	//insert/update/delete
	Action string
	// Header can be used to inspect the event
	Header *replication.EventHeader
	// contains filtered or unexported fields
}

RequestMessage 请求消息

func (*RequestMessage) Body

func (r *RequestMessage) Body() []byte

func (*RequestMessage) ColumnNames

func (r *RequestMessage) ColumnNames() []string

func (*RequestMessage) From

func (r *RequestMessage) From() string

func (*RequestMessage) GetError

func (r *RequestMessage) GetError() error

func (*RequestMessage) GetMsg

func (r *RequestMessage) GetMsg() *types.RuleMsg

func (*RequestMessage) GetParam

func (r *RequestMessage) GetParam(key string) string

func (*RequestMessage) Headers

func (r *RequestMessage) Headers() textproto.MIMEHeader

func (*RequestMessage) PKColumns

func (r *RequestMessage) PKColumns() []string

func (*RequestMessage) SetBody

func (r *RequestMessage) SetBody(body []byte)

func (*RequestMessage) SetError

func (r *RequestMessage) SetError(err error)

func (*RequestMessage) SetMsg

func (r *RequestMessage) SetMsg(msg *types.RuleMsg)

func (*RequestMessage) SetStatusCode

func (r *RequestMessage) SetStatusCode(statusCode int)

type ResponseMessage

type ResponseMessage struct {
	Table  *schema.Table
	Action string
	// contains filtered or unexported fields
}

ResponseMessage http响应消息

func (*ResponseMessage) Body

func (r *ResponseMessage) Body() []byte

func (*ResponseMessage) From

func (r *ResponseMessage) From() string

func (*ResponseMessage) GetError

func (r *ResponseMessage) GetError() error

func (*ResponseMessage) GetMsg

func (r *ResponseMessage) GetMsg() *types.RuleMsg

func (*ResponseMessage) GetParam

func (r *ResponseMessage) GetParam(key string) string

func (*ResponseMessage) Headers

func (r *ResponseMessage) Headers() textproto.MIMEHeader

func (*ResponseMessage) SetBody

func (r *ResponseMessage) SetBody(body []byte)

func (*ResponseMessage) SetError

func (r *ResponseMessage) SetError(err error)

func (*ResponseMessage) SetMsg

func (r *ResponseMessage) SetMsg(msg *types.RuleMsg)

func (*ResponseMessage) SetStatusCode

func (r *ResponseMessage) SetStatusCode(statusCode int)