streamer

Documentation

Index

• Variables
• func CollectAllBinlogFiles(dir string) ([]string, error)
• func CollectBinlogFilesCmp(dir, baseFile string, cmp FileCmp) ([]string, error)
• func ExtractPos(pos mysql.Position, uuids []string) (uuidWithSuffix string, uuidSuffix string, realPos mysql.Position, err error)
• func GetBinlogFileIndex(filename string) (float64, error)
• func RealMySQLPos(pos mysql.Position) (mysql.Position, error)
• type BinlogReader
  • func NewBinlogReader(cfg *BinlogReaderConfig) *BinlogReader
  • func (r *BinlogReader) Close() error
  • func (r *BinlogReader) StartSync(pos mysql.Position) (Streamer, error)
• type BinlogReaderConfig
• type FileCmp
• type LocalStreamer
  • func (s *LocalStreamer) GetEvent(ctx context.Context) (*replication.BinlogEvent, error)
• type ReaderHub
  • func GetReaderHub() *ReaderHub
  • func (h *ReaderHub) EarliestActiveRelayLog() *RelayLogInfo
  • func (h *ReaderHub) RemoveActiveRelayLog(taskName string)
  • func (h *ReaderHub) UpdateActiveRelayLog(taskName, uuid, filename string) error
• type RelayLogInfo
  • func (info *RelayLogInfo) Earlier(other *RelayLogInfo) bool
  • func (info *RelayLogInfo) String() string
• type Streamer

Variables

var (
	// ErrInvalidBinlogFilename means error about invalid binlog file name.
	ErrInvalidBinlogFilename = errors.New("invalid binlog file name")
	// ErrEmptyRelayDir means error about empty relay dir.
	ErrEmptyRelayDir = errors.New("empty relay dir")
)

var (
	ErrReaderRunning          = errors.New("binlog reader is already running")
	ErrBinlogFileNotSpecified = errors.New("binlog file must be specified")
)

errors used by reader

var (
	ErrNeedSyncAgain = errors.New("Last sync error or closed, try sync and get event again")
	ErrSyncClosed    = errors.New("Sync was closed")
)

errors used by streamer

Functions

func CollectAllBinlogFiles

func CollectAllBinlogFiles(dir string) ([]string, error)

CollectAllBinlogFiles collects all valid binlog files in dir

func CollectBinlogFilesCmp

func CollectBinlogFilesCmp(dir, baseFile string, cmp FileCmp) ([]string, error)

CollectBinlogFilesCmp collects valid binlog files with a compare condition

func ExtractPos

func ExtractPos(pos mysql.Position, uuids []string) (uuidWithSuffix string, uuidSuffix string, realPos mysql.Position, err error)

ExtractPos extracts (uuidWithSuffix, uuidSuffix, originalPos) from input pos (originalPos or convertedPos)

func GetBinlogFileIndex

func GetBinlogFileIndex(filename string) (float64, error)

GetBinlogFileIndex return a float64 value of index.

func RealMySQLPos

func RealMySQLPos(pos mysql.Position) (mysql.Position, error)

RealMySQLPos parses relay position and returns a mysql position and whether error occurs if parsed successfully and `UUIDSuffix` exists, sets position Name to `originalPos.NamePrefix + baseSeqSeparator + originalPos.NameSuffix`. if parsed failed returns given position and the traced error.

Types

type BinlogReader

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

BinlogReader is a binlog reader.

func NewBinlogReader

func NewBinlogReader(cfg *BinlogReaderConfig) *BinlogReader

NewBinlogReader creates a new BinlogReader

func (*BinlogReader) Close

func (r *BinlogReader) Close() error

Close closes BinlogReader.

func (*BinlogReader) StartSync

func (r *BinlogReader) StartSync(pos mysql.Position) (Streamer, error)

StartSync start syncon TODO: thread-safe?

type BinlogReaderConfig

type BinlogReaderConfig struct {
	RelayDir string
	Timezone *time.Location
}

BinlogReaderConfig is the configuration for BinlogReader

type FileCmp

type FileCmp uint8

FileCmp is a compare condition used when collecting binlog files

const (
	FileCmpLess FileCmp = iota + 1
	FileCmpLessEqual
	FileCmpEqual
	FileCmpBiggerEqual
	FileCmpBigger
)

FileCmpLess represents a < FileCmp condition, others are similar

type LocalStreamer

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

LocalStreamer reads and parses binlog events from local binlog file.

func (*LocalStreamer) GetEvent

func (s *LocalStreamer) GetEvent(ctx context.Context) (*replication.BinlogEvent, error)

GetEvent gets the binlog event one by one, it will block until parser occurs some errors. You can pass a context (like Cancel or Timeout) to break the block.

type ReaderHub

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

ReaderHub holds information for all active Readers

func GetReaderHub

func GetReaderHub() *ReaderHub

GetReaderHub gets singleton instance of ReaderHub

func (*ReaderHub) EarliestActiveRelayLog

func (h *ReaderHub) EarliestActiveRelayLog() *RelayLogInfo

EarliestActiveRelayLog implements RelayOperator.EarliestActiveRelayLog

func (*ReaderHub) RemoveActiveRelayLog

func (h *ReaderHub) RemoveActiveRelayLog(taskName string)

RemoveActiveRelayLog removes active relay log for taskName

func (*ReaderHub) UpdateActiveRelayLog

func (h *ReaderHub) UpdateActiveRelayLog(taskName, uuid, filename string) error

UpdateActiveRelayLog updates active relay log for taskName

type RelayLogInfo

type RelayLogInfo struct {
	TaskName   string
	UUID       string
	UUIDSuffix int
	Filename   string
}

RelayLogInfo represents information for relay log

func (*RelayLogInfo) Earlier

func (info *RelayLogInfo) Earlier(other *RelayLogInfo) bool

Earlier checks whether this relay log file is earlier than the other

func (*RelayLogInfo) String

func (info *RelayLogInfo) String() string

String implements Stringer.String

type Streamer

type Streamer interface {
	// GetEvent returns binlog event
	GetEvent(ctx context.Context) (*replication.BinlogEvent, error)
}

Streamer provides the ability to get binlog event from remote server or local file.