mysql_binlog

README

mysql_binlog

Installation

Run command on you [$GOPATH/src] path:

go get -u github.com/alex60217101990/mysql_binlog

Usage

[...]

Create model:

type GameLogo struct {
	ID          int64  `json:"-" gorm:"primary_key;column:id;" sql:"type:int unsigned auto_increment"`
	GameID      int64  `json:"game_id" gorm:"column:game_id;type:bigint;"`
	GameLogoURL string `json:"game_logo" gorm:"type:varchar(500);column:game_logo;"`
}

// must be implement method: [TableName() string]
func (GameLogo) TableName() string {
	return "games_logos"
}

// must be implement method: [FromBinLog(fields ...interface{}) interface{}]
func (GameLogo) FromBinLog(fields ...interface{}) interface{} {
	g := GameLogo{}
	g.ID = int64(fields[0].(int32))
	g.GameID = fields[1].(int64)
	g.GameLogoURL = fields[2].(string)
	return g
}

then:

	host := "localhost"
	port := 3307
	schema := "trade"
	user := "root"
	password := "password"
	size := uint16(200)

	binLog, err := NewBilLogService(
		SetQueueDirection(Ring),
		SetQueueSize(&size),
		SetConfig(&Config{
			Host:       &host,
			Port:       &port,
			Schema:     &schema,
			User:       &user,
			Password:   &password,
			GormModels: []DBModel{&GameLogo{}}, // or other you gorm models
		}),
	)
	if err != nil {
		log.Println(err)
	}

	go binLog.Listen()

	defer binLog.Close()

	go func() {
		for {
			select {
			case event, ok := <-binLog.Subscriber():
				if ok {
					if logEvent, ok := event.(BinLogEvent); ok {
						if logo, ok := logEvent.Value.(GameLogo); ok {
							fmt.Println(logo)
						}
						if logEvent.OldValue != nil {
							if logo, ok := logEvent.OldValue.(GameLogo); ok {
								fmt.Println(logo)
							}
						}
					}
				}
			}
		}
	}()

[...]

'BinLogEvent' has signature:

[...]

type BinLogEvent struct {
	Type     ActionType
	Value    interface{}
	OldValue interface{}
}

[...]

License

MIT

Documentation

Index

• Constants
• func SetConfig(configs *Config) func(*BinLogService) error
• func SetLogger(logger *log.Logger) func(*BinLogService) error
• func SetQueueDirection(queueDirection QueueDirection) func(*BinLogService) error
• func SetQueueSize(size *uint16) func(*BinLogService) error
• type ActionType
  • func (t ActionType) String() string
  • func (t ActionType) Val() int8
• type BinLogEvent
• type BinLogService
  • func NewBilLogService(options ...func(*BinLogService) error) (*BinLogService, error)
  • func (b *BinLogService) Close()
  • func (b *BinLogService) Listen()
  • func (b *BinLogService) Subscriber() <-chan interface{}
• type BinlogParser
  • func (m *BinlogParser) GetBinLogData(element interface{}, e *canal.RowsEvent, n int) error
• type Config
• type DBModel
• type GameLogo
  • func (GameLogo) FromBinLog(fields ...interface{}) interface{}
  • func (GameLogo) TableName() string
• type QueueDirection
  • func (t QueueDirection) String() string
  • func (t QueueDirection) Val() int8

Constants

const (
	DefaultActionChanSize uint16 = 500
)

Functions

func SetConfig

func SetConfig(configs *Config) func(*BinLogService) error

func SetLogger

func SetLogger(logger *log.Logger) func(*BinLogService) error

func SetQueueDirection

func SetQueueDirection(queueDirection QueueDirection) func(*BinLogService) error

func SetQueueSize

func SetQueueSize(size *uint16) func(*BinLogService) error

Types

type ActionType

type ActionType int8

const (
	Insert ActionType = (1 + iota)
	Update
	Delete
)

func (ActionType) String

func (t ActionType) String() string

func (ActionType) Val

func (t ActionType) Val() int8

type BinLogEvent

type BinLogEvent struct {
	Type     ActionType
	Value    interface{}
	OldValue interface{}
}

type BinLogService

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

func NewBilLogService

func NewBilLogService(options ...func(*BinLogService) error) (*BinLogService, error)

func (*BinLogService) Close

func (b *BinLogService) Close()

func (*BinLogService) Listen

func (b *BinLogService) Listen()

func (*BinLogService) Subscriber

func (b *BinLogService) Subscriber() <-chan interface{}

type BinlogParser

type BinlogParser struct{}

func (*BinlogParser) GetBinLogData

func (m *BinlogParser) GetBinLogData(element interface{}, e *canal.RowsEvent, n int) error

type Config

type Config struct {
	Host       *string
	Port       *int
	Schema     *string
	User       *string
	Password   *string
	GormModels []DBModel
}

type DBModel

type DBModel interface {
	TableName() string
	FromBinLog(fields ...interface{}) interface{}
}

type GameLogo

type GameLogo struct {
	ID          int64  `json:"-" gorm:"primary_key;column:id;" sql:"type:int unsigned auto_increment"`
	GameID      int64  `json:"game_id" gorm:"column:game_id;type:bigint;"`
	GameLogoURL string `json:"game_logo" gorm:"type:varchar(500);column:game_logo;"`
}

func (GameLogo) FromBinLog

func (GameLogo) FromBinLog(fields ...interface{}) interface{}

func (GameLogo) TableName

func (GameLogo) TableName() string

type QueueDirection

type QueueDirection int8

const (
	Ring QueueDirection = (1 + iota)
	Overflowing
)

func (QueueDirection) String

func (t QueueDirection) String() string

func (QueueDirection) Val

func (t QueueDirection) Val() int8