src

Documentation

Overview

日志模式同步 update 转成 insert on update insert 转成 replace into delete 转成 insert on update 只要是同一条数据，只要有遍历过，后面遍历出来的数据，则不再进行操作

普通模式同步 update 转成 insert on update insert 转成 replace into delete 转成 delete 只要是同一条数据，只要有遍历过，后面遍历出来的数据，则不再进行操作

Index

• Constants
• func NewConn() pluginDriver.Driver
• func NewMysqlDBConn(uri string) *mysqlDB
• type Conn
  • func (This *Conn) AutoCommit() (LastSuccessCommitData *pluginDriver.PluginDataType, ...)
  • func (This *Conn) AutoTableCommit(list []*pluginDriver.PluginDataType) (ErrData *pluginDriver.PluginDataType, e error)
  • func (This *Conn) CheckDataSkip(data *pluginDriver.PluginDataType) bool
  • func (This *Conn) CheckUri() error
  • func (This *Conn) Close() bool
  • func (This *Conn) Commit(data *pluginDriver.PluginDataType, retry bool) (*pluginDriver.PluginDataType, *pluginDriver.PluginDataType, error)
  • func (This *Conn) CommitLogMod_Append(list []*pluginDriver.PluginDataType) (errData *pluginDriver.PluginDataType)
  • func (This *Conn) CommitLogMod_Update(list []*pluginDriver.PluginDataType) (errData *pluginDriver.PluginDataType)
  • func (This *Conn) CommitNormal(list []*pluginDriver.PluginDataType) (errData *pluginDriver.PluginDataType)
  • func (This *Conn) Connect() bool
  • func (This *Conn) Del(data *pluginDriver.PluginDataType, retry bool) (*pluginDriver.PluginDataType, *pluginDriver.PluginDataType, error)
  • func (This *Conn) GetParam(p interface{}) (*PluginParam, error)
  • func (This *Conn) GetUriExample() string
  • func (This *Conn) Insert(data *pluginDriver.PluginDataType, retry bool) (*pluginDriver.PluginDataType, *pluginDriver.PluginDataType, error)
  • func (This *Conn) NotAutoTableCommit(list []*pluginDriver.PluginDataType) (ErrData *pluginDriver.PluginDataType, e error)
  • func (This *Conn) Open() error
  • func (This *Conn) Query(data *pluginDriver.PluginDataType, retry bool) (LastSuccessCommitData *pluginDriver.PluginDataType, ...)
  • func (This *Conn) ReConnect() bool
  • func (This *Conn) ReplaceTwoReplace(sql string) string
  • func (This *Conn) SetOption(uri *string, param map[string]interface{})
  • func (This *Conn) SetParam(p interface{}) (interface{}, error)
  • func (This *Conn) Skip(SkipData *pluginDriver.PluginDataType) error
  • func (This *Conn) StmtClose()
  • func (This *Conn) TimeOutCommit() (LastSuccessCommitData *pluginDriver.PluginDataType, ...)
  • func (This *Conn) TranferDMLSql(data *pluginDriver.PluginDataType) (newSqlArr []string)
  • func (This *Conn) TranferQuerySql(data *pluginDriver.PluginDataType) (newSqlArr []string)
  • func (This *Conn) TransferNotes2Space(sql string) string
  • func (This *Conn) Update(data *pluginDriver.PluginDataType, retry bool) (*pluginDriver.PluginDataType, *pluginDriver.PluginDataType, error)
• type EventType
• type PluginMySQLController
  • func (c *PluginMySQLController) GetMysqlSchemaList()
  • func (c *PluginMySQLController) GetMysqlSchemaTableList()
  • func (c *PluginMySQLController) GetMysqlTableFields()
• type PluginParam
• type PluginParam0
• type SyncMode
• type TableDataStruct
  • func NewTableData() *TableDataStruct
• type TableStruct

Constants

const BIFROST_VERION = "v1.7.3"

const VERSION = "v1.7.3"

Functions

func NewConn

func NewConn() pluginDriver.Driver

func NewMysqlDBConn

func NewMysqlDBConn(uri string) *mysqlDB

Types

type Conn

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

func (*Conn) AutoCommit

func (This *Conn) AutoCommit() (LastSuccessCommitData *pluginDriver.PluginDataType, ErrData *pluginDriver.PluginDataType, e error)

func (*Conn) AutoTableCommit

func (This *Conn) AutoTableCommit(list []*pluginDriver.PluginDataType) (ErrData *pluginDriver.PluginDataType, e error)

自动创建表的提交

func (*Conn) CheckDataSkip

func (This *Conn) CheckDataSkip(data *pluginDriver.PluginDataType) bool

func (*Conn) CheckUri

func (This *Conn) CheckUri() error

func (*Conn) Close

func (This *Conn) Close() bool

func (*Conn) Commit

func (This *Conn) Commit(data *pluginDriver.PluginDataType, retry bool) (*pluginDriver.PluginDataType, *pluginDriver.PluginDataType, error)

func (*Conn) CommitLogMod_Append

func (This *Conn) CommitLogMod_Append(list []*pluginDriver.PluginDataType) (errData *pluginDriver.PluginDataType)

func (*Conn) CommitLogMod_Update

func (This *Conn) CommitLogMod_Update(list []*pluginDriver.PluginDataType) (errData *pluginDriver.PluginDataType)

func (*Conn) CommitNormal

func (This *Conn) CommitNormal(list []*pluginDriver.PluginDataType) (errData *pluginDriver.PluginDataType)

func (*Conn) Connect

func (This *Conn) Connect() bool

func (*Conn) Del

func (This *Conn) Del(data *pluginDriver.PluginDataType, retry bool) (*pluginDriver.PluginDataType, *pluginDriver.PluginDataType, error)

func (*Conn) GetParam

func (This *Conn) GetParam(p interface{}) (*PluginParam, error)

func (*Conn) GetUriExample

func (This *Conn) GetUriExample() string

func (*Conn) Insert

func (This *Conn) Insert(data *pluginDriver.PluginDataType, retry bool) (*pluginDriver.PluginDataType, *pluginDriver.PluginDataType, error)

func (*Conn) NotAutoTableCommit

func (This *Conn) NotAutoTableCommit(list []*pluginDriver.PluginDataType) (ErrData *pluginDriver.PluginDataType, e error)

func (*Conn) Open

func (This *Conn) Open() error

func (*Conn) Query

func (This *Conn) Query(data *pluginDriver.PluginDataType, retry bool) (LastSuccessCommitData *pluginDriver.PluginDataType, ErrData *pluginDriver.PluginDataType, err error)

func (*Conn) ReConnect

func (This *Conn) ReConnect() bool

func (*Conn) ReplaceTwoReplace

func (This *Conn) ReplaceTwoReplace(sql string) string

去除连续的两个空格

func (*Conn) SetOption

func (This *Conn) SetOption(uri *string, param map[string]interface{})

func (*Conn) SetParam

func (This *Conn) SetParam(p interface{}) (interface{}, error)

func (*Conn) Skip

func (This *Conn) Skip(SkipData *pluginDriver.PluginDataType) error

设置跳过的位点

func (*Conn) StmtClose

func (This *Conn) StmtClose()

func (*Conn) TimeOutCommit

func (This *Conn) TimeOutCommit() (LastSuccessCommitData *pluginDriver.PluginDataType, ErrData *pluginDriver.PluginDataType, err error)

func (*Conn) TranferDMLSql

func (This *Conn) TranferDMLSql(data *pluginDriver.PluginDataType) (newSqlArr []string)

func (*Conn) TranferQuerySql

func (This *Conn) TranferQuerySql(data *pluginDriver.PluginDataType) (newSqlArr []string)

func (*Conn) TransferNotes2Space

func (This *Conn) TransferNotes2Space(sql string) string

func (*Conn) Update

func (This *Conn) Update(data *pluginDriver.PluginDataType, retry bool) (*pluginDriver.PluginDataType, *pluginDriver.PluginDataType, error)

type EventType

type EventType int8

const (
	INSERT         EventType = 0
	UPDATE         EventType = 1
	DELETE         EventType = 2
	REPLACE_INSERT EventType = 3
	SQLTYPE        EventType = 4
)

type PluginMySQLController

type PluginMySQLController struct {
	controller.CommonController
}

func (*PluginMySQLController) GetMysqlSchemaList

func (c *PluginMySQLController) GetMysqlSchemaList()

func (*PluginMySQLController) GetMysqlSchemaTableList

func (c *PluginMySQLController) GetMysqlSchemaTableList()

func (*PluginMySQLController) GetMysqlTableFields

func (c *PluginMySQLController) GetMysqlTableFields()

type PluginParam

type PluginParam struct {
	Field                []fieldStruct
	BatchSize            int
	Schema               string
	Table                string
	NullTransferDefault  bool //是否将null值强制转成相对应类型的默认值
	SyncMode             SyncMode
	BifrostMustBeSuccess bool // bifrost server 保留,数据是否能丢

	PriKey []fieldStruct

	Data *TableDataStruct

	AutoTable bool // 是否自动匹配数据表

	SkipBinlogData *pluginDriver.PluginDataType // 在执行 skip 的时候 ，进行传入进来的时候需要要过滤的 位点，在每次commit之后，这个数据会被清空
	// contains filtered or unexported fields
}

type PluginParam0

type PluginParam0 struct {
	Field          []fieldStruct
	SchemaName     string
	TableName      string
	SchemaAndTable string
	PriKey         []fieldStruct // 主键对应关系
	FromPriKey     string        // 源表的 主键 字段
	ToPriKey       string        // 目标库的 主键 字段
}

type SyncMode

type SyncMode string

const (
	SYNCMODE_NORMAL       SyncMode = "Normal"
	SYNCMODE_LOG_UPDATE   SyncMode = "LogUpdate"
	SYNCMODE_LOG_APPEND   SyncMode = "LogAppend"
	SYNCMODE_NO_SYNC_DATA SyncMode = "NoSyncData"
)

type TableDataStruct

type TableDataStruct struct {
	Data       []*pluginDriver.PluginDataType
	CommitData []*pluginDriver.PluginDataType // commit 提交的数据列表，Data 每 BatchSize 数据量划分为一个最后提交的commit
}

func NewTableData

func NewTableData() *TableDataStruct

type TableStruct

type TableStruct struct {
	COLUMN_NAME       string
	COLUMN_DEFAULT    *string
	IS_NULLABLE       string
	COLUMN_TYPE       string
	COLUMN_KEY        string
	EXTRA             string
	COLUMN_COMMENT    string
	DATA_TYPE         string
	NUMERIC_PRECISION *uint64
	NUMERIC_SCALE     *uint64
}