model

Documentation

Index

• func ServiceNames(services []Service) []string
• type Abbreviater
• type Config
• type DDLRequest
• type Destination
• type Duration
  • func (r *Duration) UnmarshalYAML(data []byte) error
• type Handler
  • func NewHandler(queue chan interface{}, syncMarkTable, syncChannel string, ...) *Handler
  • func (s *Handler) OnDDL(timestamp uint32, _ mysql.Position, e *replication.QueryEvent, stmt string) error
  • func (s *Handler) OnPosSynced(timestamp uint32, nextPos mysql.Position, _ mysql.GTIDSet, force bool) error
  • func (s *Handler) OnRow(timestamp uint32, e *calan.RowsEvent) error
  • func (s *Handler) OnXID(_ uint32, _ mysql.Position) error
  • func (s *Handler) String() string
• type PosRequest
• type RawBytes
  • func (r RawBytes) Len() int
  • func (r *RawBytes) UnmarshalYAML(data []byte) error
• type Row
  • func (r *Row) Abbreviate() Row
• type RowsRequest
  • func (r RowsRequest) Abbreviate() RowsRequest
  • func (r *RowsRequest) FixBytesValues()
• type RowsRequests
  • func (r RowsRequests) Abbreviate() interface{}
• type Service
  • func (s Service) Close()
  • func (s Service) Start(ctx context.Context, handler *http.ServeMux) error
• type Source
• type Starter

Functions

func ServiceNames

func ServiceNames(services []Service) []string

Types

type Abbreviater

type Abbreviater interface {
	Abbreviate() interface{}
}

type Config

type Config struct {
	Web      string
	Services []Service
}

type DDLRequest

type DDLRequest struct {
	Timestamp uint32 // binlog发生时间，unix毫秒数的时间戳
	Schema    string // 当前schema（即当前所在数据库的库名)
	Stmt      string // DDL语句
}

DDLRequest 是数据定义请求.

type Destination

type Destination interface {
	Write(ctx context.Context, v interface{}) error
}

type Duration

type Duration time.Duration

func (*Duration) UnmarshalYAML

func (r *Duration) UnmarshalYAML(data []byte) error

type Handler

type Handler struct {
	calan.DummyEventHandler
	// contains filtered or unexported fields
}

func NewHandler

func NewHandler(queue chan interface{}, syncMarkTable, syncChannel string,
	posExtraCallback func(p interface{})) *Handler

func (*Handler) OnDDL

func (s *Handler) OnDDL(timestamp uint32, _ mysql.Position, e *replication.QueryEvent, stmt string) error

func (*Handler) OnPosSynced

func (s *Handler) OnPosSynced(timestamp uint32, nextPos mysql.Position, _ mysql.GTIDSet, force bool) error

func (*Handler) OnRow

func (s *Handler) OnRow(timestamp uint32, e *calan.RowsEvent) error

func (*Handler) OnXID

func (s *Handler) OnXID(_ uint32, _ mysql.Position) error

OnXID commits a transaction.

func (*Handler) String

func (s *Handler) String() string

type PosRequest

type PosRequest struct {
	Timestamp uint32
	Name      string
	Pos       uint32
}

PosRequest 是binlog位置更新请求.

type RawBytes

type RawBytes []byte

func (RawBytes) Len

func (r RawBytes) Len() int

func (*RawBytes) UnmarshalYAML

func (r *RawBytes) UnmarshalYAML(data []byte) error

type Row

type Row struct {
	Old []interface{} `json:",omitempty"` // 更新之前的取值（只有update时有)
	Row []interface{} // 更新后的取值
}

Row 更新行.

func (*Row) Abbreviate

func (r *Row) Abbreviate() Row

type RowsRequest

type RowsRequest struct {
	Timestamp    uint32       // binlog发生时间，unix毫秒数的时间戳
	Schema       string       // 数据更新的schema（即数据库的库名)
	Table        string       // 数据更新的表名
	Action       calan.Action // 更新动作: update/insert/delete
	Columns      []string     // 更新字段列表
	Pks          []int        // 主键字段列表
	BytesColumns []int        // 是[]byte类型的字段，json序列化后是base64，反序列化会成为string，所以需要标记一下
	PkColumns    []string     // 主键字段列表
	Rows         []Row        // 更新行
}

RowsRequest 是数据更新请求.

func (RowsRequest) Abbreviate

func (r RowsRequest) Abbreviate() RowsRequest

func (*RowsRequest) FixBytesValues

func (r *RowsRequest) FixBytesValues()

type RowsRequests

type RowsRequests struct {
	Requests []RowsRequest
}

func (RowsRequests) Abbreviate

func (r RowsRequests) Abbreviate() interface{}

type Service

type Service struct {
	Name string
	Source
	Destination
}

func (Service) Close

func (s Service) Close()

func (Service) Start

func (s Service) Start(ctx context.Context, handler *http.ServeMux) error

type Source

type Source interface {
	ReadChan() chan interface{}
}

type Starter

type Starter interface {
	Start(ctx context.Context, handler *http.ServeMux) error
}