driver

Documentation

Index

• Constants
• func CheckUri(name string, uri *string) error
• func DeepCopy(dst, src interface{}) error
• func Drivers() map[string]DriverStructure
• func GetApiVersion() string
• func Register(name string, NewDriverFun NewDriver, version string, bifrost_version string)
• func TransfeResult(val string, data *PluginDataType, rowIndex int) interface{}
• type ConnStatus
• type Driver
  • func Open(name string, uri *string) Driver
• type DriverStructure
• type NewDriver
• type PluginDataType
  • func (c *PluginDataType) MarshalJSON() (b []byte, err error)
  • func (c *PluginDataType) UnmarshalJSON(data []byte) error
• type PluginDataTypeCopy
• type PluginDriverInterface
  • func (c *PluginDriverInterface) CheckUri() error
  • func (c *PluginDriverInterface) Close() bool
  • func (c *PluginDriverInterface) Commit(data *PluginDataType, retry bool) (LastSuccessCommitData *PluginDataType, ErrData *PluginDataType, err error)
  • func (c *PluginDriverInterface) Del(data *PluginDataType, retry bool) (LastSuccessCommitData *PluginDataType, ErrData *PluginDataType, err error)
  • func (c *PluginDriverInterface) GetUriExample() string
  • func (c *PluginDriverInterface) Insert(data *PluginDataType, retry bool) (LastSuccessCommitData *PluginDataType, ErrData *PluginDataType, err error)
  • func (c *PluginDriverInterface) Open() error
  • func (c *PluginDriverInterface) Query(data *PluginDataType, retry bool) (LastSuccessCommitData *PluginDataType, ErrData *PluginDataType, err error)
  • func (c *PluginDriverInterface) SetOption(uri *string, param map[string]interface{})
  • func (c *PluginDriverInterface) SetParam(p interface{}) (interface{}, error)
  • func (c *PluginDriverInterface) Skip(SkipData *PluginDataType) error
  • func (c *PluginDriverInterface) TimeOutCommit() (LastSuccessCommitData *PluginDataType, ErrData *PluginDataType, err error)
  • func (c *PluginDriverInterface) Update(data *PluginDataType, retry bool) (LastSuccessCommitData *PluginDataType, ErrData *PluginDataType, err error)

Constants

const API_VERSION = "v1.6"

const RegularxEpression = `\{\$([a-zA-Z0-9\-\_\[\]\'\"]+)\}`

const RegularxEpressionKey = `([a-zA-Z0-9\-\_]+)`

Functions

func CheckUri

func CheckUri(name string, uri *string) error

func DeepCopy

func DeepCopy(dst, src interface{}) error

深度拷贝对象

func Drivers

func Drivers() map[string]DriverStructure

func GetApiVersion

func GetApiVersion() string

func Register

func Register(name string, NewDriverFun NewDriver, version string, bifrost_version string)

func TransfeResult

func TransfeResult(val string, data *PluginDataType, rowIndex int) interface{}

Types

type ConnStatus

type ConnStatus int8

const (
	CLOSED  ConnStatus = 0
	STOPPED ConnStatus = 1
	RUNNING ConnStatus = 2
)

type Driver

type Driver interface {
	SetOption(uri *string, param map[string]interface{})
	Open() error
	Close() bool
	GetUriExample() string
	CheckUri() error
	Insert(data *PluginDataType, retry bool) (*PluginDataType, *PluginDataType, error) //binlog位点，处理了多少条数据,错误信息
	Update(data *PluginDataType, retry bool) (*PluginDataType, *PluginDataType, error)
	Del(data *PluginDataType, retry bool) (*PluginDataType, *PluginDataType, error)
	Query(data *PluginDataType, retry bool) (*PluginDataType, *PluginDataType, error)
	Commit(data *PluginDataType, retry bool) (*PluginDataType, *PluginDataType, error)
	SetParam(p interface{}) (interface{}, error)
	TimeOutCommit() (*PluginDataType, *PluginDataType, error)
	Skip(*PluginDataType) error
}

func Open

func Open(name string, uri *string) Driver

type DriverStructure

type DriverStructure struct {
	Version        string // 插件版本
	BifrostVersion string // 插件开发所使用的Bifrost的版本
	Error          string
	ExampleConnUri string
	// contains filtered or unexported fields
}

type NewDriver

type NewDriver func() Driver

type PluginDataType

type PluginDataType struct {
	Timestamp      uint32
	EventType      string
	Rows           []map[string]interface{}
	Query          string
	SchemaName     string
	TableName      string
	BinlogFileNum  int
	BinlogPosition uint32
	Gtid           string
	Pri            []string
	EventID        uint64
	ColumnMapping  map[string]string
}

func (*PluginDataType) MarshalJSON

func (c *PluginDataType) MarshalJSON() (b []byte, err error)

deep copy data, then int32,uint64... to string 这里要进行 deepcopy 是防止同时多个地方在使用同一份数据

func (*PluginDataType) UnmarshalJSON

func (c *PluginDataType) UnmarshalJSON(data []byte) error

MarshalJSON 的时候，将 int32,uint64... to string 了 这里再将 对应的 string 转成 int32,uint64...

type PluginDataTypeCopy

type PluginDataTypeCopy PluginDataType

type PluginDriverInterface

type PluginDriverInterface struct {
}

func (*PluginDriverInterface) CheckUri

func (c *PluginDriverInterface) CheckUri() error

func (*PluginDriverInterface) Close

func (c *PluginDriverInterface) Close() bool

func (*PluginDriverInterface) Commit

func (c *PluginDriverInterface) Commit(data *PluginDataType, retry bool) (LastSuccessCommitData *PluginDataType, ErrData *PluginDataType, err error)

func (*PluginDriverInterface) Del

func (c *PluginDriverInterface) Del(data *PluginDataType, retry bool) (LastSuccessCommitData *PluginDataType, ErrData *PluginDataType, err error)

func (*PluginDriverInterface) GetUriExample

func (c *PluginDriverInterface) GetUriExample() string

func (*PluginDriverInterface) Insert

func (c *PluginDriverInterface) Insert(data *PluginDataType, retry bool) (LastSuccessCommitData *PluginDataType, ErrData *PluginDataType, err error)

func (*PluginDriverInterface) Open

func (c *PluginDriverInterface) Open() error

func (*PluginDriverInterface) Query

func (c *PluginDriverInterface) Query(data *PluginDataType, retry bool) (LastSuccessCommitData *PluginDataType, ErrData *PluginDataType, err error)

func (*PluginDriverInterface) SetOption

func (c *PluginDriverInterface) SetOption(uri *string, param map[string]interface{})

func (*PluginDriverInterface) SetParam

func (c *PluginDriverInterface) SetParam(p interface{}) (interface{}, error)

func (*PluginDriverInterface) Skip

func (c *PluginDriverInterface) Skip(SkipData *PluginDataType) error

func (*PluginDriverInterface) TimeOutCommit

func (c *PluginDriverInterface) TimeOutCommit() (LastSuccessCommitData *PluginDataType, ErrData *PluginDataType, err error)

func (*PluginDriverInterface) Update

func (c *PluginDriverInterface) Update(data *PluginDataType, retry bool) (LastSuccessCommitData *PluginDataType, ErrData *PluginDataType, err error)