simple

Documentation

Index

• func NewBuilder(ctx context.Context, config *common.Config) (*builder, error)
• type Decoder
  • func NewDecoder(ctx context.Context, config *common.Config, db *sql.DB) (*Decoder, error)
  • func (d *Decoder) AddKeyValue(_, value []byte) (err error)
  • func (d *Decoder) GetCachedEvents() []*model.RowChangedEvent
  • func (d *Decoder) HasNext() (model.MessageType, bool, error)
  • func (d *Decoder) NextDDLEvent() (*model.DDLEvent, error)
  • func (d *Decoder) NextResolvedEvent() (uint64, error)
  • func (d *Decoder) NextRowChangedEvent() (*model.RowChangedEvent, error)
• type IndexSchema
• type MessageType
• type TableInfoProvider
• type TableSchema

Functions

func NewBuilder

func NewBuilder(ctx context.Context, config *common.Config) (*builder, error)

NewBuilder returns a new builder

Types

type Decoder

type Decoder struct {

	// CachedRowChangedEvents are events just decoded from the cachedMessages
	CachedRowChangedEvents []*model.RowChangedEvent
	// contains filtered or unexported fields
}

Decoder implement the RowEventDecoder interface

func NewDecoder

func NewDecoder(ctx context.Context, config *common.Config, db *sql.DB) (*Decoder, error)

NewDecoder returns a new Decoder

func (*Decoder) AddKeyValue

func (d *Decoder) AddKeyValue(_, value []byte) (err error)

AddKeyValue add the received key and values to the Decoder,

func (*Decoder) GetCachedEvents

func (d *Decoder) GetCachedEvents() []*model.RowChangedEvent

GetCachedEvents returns the cached events

func (*Decoder) HasNext

func (d *Decoder) HasNext() (model.MessageType, bool, error)

HasNext returns whether there is any event need to be consumed

func (*Decoder) NextDDLEvent

func (d *Decoder) NextDDLEvent() (*model.DDLEvent, error)

NextDDLEvent returns the next DDL event if exists

func (*Decoder) NextResolvedEvent

func (d *Decoder) NextResolvedEvent() (uint64, error)

NextResolvedEvent returns the next resolved event if exists

func (*Decoder) NextRowChangedEvent

func (d *Decoder) NextRowChangedEvent() (*model.RowChangedEvent, error)

NextRowChangedEvent returns the next row changed event if exists

type IndexSchema

type IndexSchema struct {
	Name     string   `json:"name"`
	Unique   bool     `json:"unique"`
	Primary  bool     `json:"primary"`
	Nullable bool     `json:"nullable"`
	Columns  []string `json:"columns"`
}

IndexSchema is the schema of the index.

type MessageType

type MessageType string

MessageType is the type of the message.

const (
	// MessageTypeWatermark is the type of the watermark event.
	MessageTypeWatermark MessageType = "WATERMARK"
	// MessageTypeBootstrap is the type of the bootstrap event.
	MessageTypeBootstrap MessageType = "BOOTSTRAP"
	// MessageTypeDDL is the type of the ddl event.
	MessageTypeDDL MessageType = "DDL"
	// MessageTypeDML is the type of the row event.
	MessageTypeDML MessageType = "DML"
)

const (
	// DMLTypeInsert is the type of the insert event.
	DMLTypeInsert MessageType = "INSERT"
	// DMLTypeUpdate is the type of the update event.
	DMLTypeUpdate MessageType = "UPDATE"
	// DMLTypeDelete is the type of the delete event.
	DMLTypeDelete MessageType = "DELETE"
)

DML Message types

const (
	DDLTypeCreate   MessageType = "CREATE"
	DDLTypeRename   MessageType = "RENAME"
	DDLTypeCIndex   MessageType = "CINDEX"
	DDLTypeDIndex   MessageType = "DINDEX"
	DDLTypeErase    MessageType = "ERASE"
	DDLTypeTruncate MessageType = "TRUNCATE"
	DDLTypeAlter    MessageType = "ALTER"
	DDLTypeQuery    MessageType = "QUERY"
)

DDL message types

type TableInfoProvider

type TableInfoProvider interface {
	Write(info *model.TableInfo)
	Read(schema, table string, version uint64) *model.TableInfo
}

TableInfoProvider is used to store and read table info It works like a schema cache when consuming simple protocol messages It will store multiple versions of table info for a table The table info which has the exact (schema, table, version) will be returned when reading

type TableSchema

type TableSchema struct {
	Schema  string          `json:"schema"`
	Table   string          `json:"table"`
	TableID int64           `json:"tableID"`
	Version uint64          `json:"version"`
	Columns []*columnSchema `json:"columns"`
	Indexes []*IndexSchema  `json:"indexes"`
}

TableSchema is the schema of the table.