relay

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

type BinlogReaderConfig struct {
	RelayDir            string
	Timezone            *time.Location
	Flavor              string
	RowsEventDecodeFunc func(*replication.RowsEvent, []byte) error
}

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

type BinlogWriterStatus struct {
	Filename string `json:"filename"`
	Offset   int64  `json:"offset"`
}

type Config struct {
	EnableGTID bool `toml:"enable-gtid" json:"enable-gtid"`
	// deprecated
	AutoFixGTID bool              `toml:"auto-fix-gtid" json:"auto-fix-gtid"`
	RelayDir    string            `toml:"relay-dir" json:"relay-dir"`
	ServerID    uint32            `toml:"server-id" json:"server-id"`
	Flavor      string            `toml:"flavor" json:"flavor"`
	Charset     string            `toml:"charset" json:"charset"`
	From        dbconfig.DBConfig `toml:"data-source" json:"data-source"`

	// synchronous start point (if no meta saved before)
	// do not need to specify binlog-pos, because relay will fetch the whole file
	BinLogName string `toml:"binlog-name" json:"binlog-name"`
	BinlogGTID string `toml:"binlog-gtid" json:"binlog-gtid"`
	UUIDSuffix int    `toml:"-" json:"-"`

	// for binlog reader retry
	ReaderRetry ReaderRetryConfig `toml:"reader-retry" json:"reader-retry"`
}

type EventNotifier interface {
	// Notified returns a channel used to check whether there is new binlog event written to the file
	Notified() chan interface{}
}

type FileCmp uint8

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

type Listener interface {
	// OnEvent get called when relay processed an event successfully.
	OnEvent(e *replication.BinlogEvent)
}

type LocalMeta struct {
	sync.RWMutex

	BinLogName string `toml:"binlog-name" json:"binlog-name"`
	BinLogPos  uint32 `toml:"binlog-pos" json:"binlog-pos"`
	BinlogGTID string `toml:"binlog-gtid" json:"binlog-gtid"`
	// contains filtered or unexported fields
}

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

type Meta interface {
	// Load loads meta information for the recently active server
	Load() error

	// AdjustWithStartPos adjusts current pos / GTID with start pos
	// if current pos / GTID is meaningless, update to start pos or last pos when start pos is meaningless
	// else do nothing
	AdjustWithStartPos(binlogName string, binlogGTID string, enableGTID bool, latestBinlogName string, latestBinlogGTID string) (bool, error)

	// Save saves meta information
	Save(pos mysql.Position, gset mysql.GTIDSet) error

	// Flush flushes meta information
	Flush() error

	// Dirty checks whether meta in memory is dirty (need to Flush)
	Dirty() bool

	// AddDir adds relay log subdirectory for a new server. The name of new subdirectory
	// consists of the server_uuid of new server and a suffix.
	// if suffix is not zero value, add sub relay directory with suffix (bound to a new source)
	// otherwise the added sub relay directory's suffix is incremented (master/slave switch)
	// after sub relay directory added, the internal binlog pos should be reset
	// and binlog pos will be set again when new binlog events received
	// if set @newPos / @newGTID, old value will be replaced
	AddDir(serverUUID string, newPos *mysql.Position, newGTID mysql.GTIDSet, suffix int) error

	// Pos returns current (UUID with suffix, Position) pair
	Pos() (string, mysql.Position)

	// GTID returns current (UUID with suffix, GTID) pair
	GTID() (string, mysql.GTIDSet)

	// SubDir returns the name of current relay log subdirectory.
	SubDir() string

	// TrimUUIDIndexFile trim invalid relay log subdirectories from memory and update the server-uuid.index file
	// return trimmed result.
	TrimUUIDIndexFile() ([]string, error)

	// Dir returns the full path of relay log subdirectory.
	Dir() string

	// String returns string representation of current meta info
	String() string
}

type Operator interface {
	// EarliestActiveRelayLog returns the earliest active relay log info in this operator
	EarliestActiveRelayLog() *streamer.RelayLogInfo
}

type Process interface {
	// Init initial relat log unit
	Init(ctx context.Context) (err error)
	// Process run background logic of relay log unit
	Process(ctx context.Context) pb.ProcessResult
	// ActiveRelayLog returns the earliest active relay log info in this operator
	ActiveRelayLog() *pkgstreamer.RelayLogInfo
	// Reload reloads config
	Reload(newCfg *Config) error
	// Update updates config
	Update(cfg *config.SubTaskConfig) error
	// Resume resumes paused relay log process unit
	Resume(ctx context.Context, pr chan pb.ProcessResult)
	// Pause pauses a running relay log process unit
	Pause()
	// Error returns error message if having one
	Error() interface{}
	// Status returns status of relay log process unit.
	Status(sourceStatus *binlog.SourceStatus) interface{}
	// Close does some clean works
	Close()
	// IsClosed returns whether relay log process unit was closed
	IsClosed() bool
	// SaveMeta save relay meta
	SaveMeta(pos mysql.Position, gset mysql.GTIDSet) error
	// ResetMeta reset relay meta
	ResetMeta()
	// PurgeRelayDir will clear all contents under w.cfg.RelayDir
	PurgeRelayDir() error
	// RegisterListener registers a relay listener
	RegisterListener(el Listener)
	// UnRegisterListener unregisters a relay listener
	UnRegisterListener(el Listener)
	// NewReader creates a new relay reader
	NewReader(logger log.Logger, cfg *BinlogReaderConfig) *BinlogReader
	// IsActive check whether given uuid+filename is active binlog file, if true return current file offset
	IsActive(uuid, filename string) (bool, int64)
}

type PurgeInterceptor interface {
	// ForbidPurge returns whether forbidding purge currently and an optional message
	ForbidPurge() (bool, string)
}

type PurgeStrategy interface {
	// Check checks whether need to do the purge in the background automatically
	Check(args interface{}) (bool, error)

	// Do does the purge process one time
	Do(args interface{}) error

	// Purging indicates whether is doing purge
	Purging() bool

	// Type returns the strategy type
	Type() strategyType
}

type Purger interface {
	// Start starts strategies by config
	Start()
	// Close stops the started strategies
	Close()
	// Purging returns whether the purger is purging
	Purging() bool
	// Do does the purge process one time
	Do(ctx context.Context, req *pb.PurgeRelayRequest) error
}

type RConfig struct {
	SyncConfig replication.BinlogSyncerConfig
	Pos        mysql.Position
	GTIDs      mysql.GTIDSet
	EnableGTID bool
	MasterID   string // the identifier for the master, used when logging.
}

type RResult struct {
	Event *replication.BinlogEvent
}

type Reader interface {
	// Start starts the reading process.
	Start() error

	// Close closes the reader and release the resource.
	Close() error

	// GetEvent gets the binlog event one by one, it will block if no event can be read.
	// You can pass a context (like Cancel) to break the block.
	GetEvent(ctx context.Context) (RResult, error)
}

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

type ReaderRetryConfig struct {
	BackoffRollback time.Duration `toml:"backoff-rollback" json:"backoff-rollback"`
	BackoffMax      time.Duration `toml:"backoff-max" json:"backoff-max"`
	// unexposed config
	BackoffMin    time.Duration `json:"-"`
	BackoffJitter bool          `json:"-"`
	BackoffFactor float64       `json:"-"`
}

type Relay struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

type StrategyArgs interface {
	// SetActiveRelayLog sets active relay log info in args
	// this should be called before do the purging
	SetActiveRelayLog(active *streamer.RelayLogInfo)
}

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

type WResult struct {
	Ignore       bool   // whether the event ignored by the writer
	IgnoreReason string // why the writer ignore the event
}

type Writer interface {
	// Init inits the writer, should be called before any other method
	Init(uuid, filename string)
	// Close closes the writer and release the resource.
	Close() error

	// WriteEvent writes an binlog event's data into disk or any other places.
	// It is not safe for concurrent use by multiple goroutines.
	WriteEvent(ev *replication.BinlogEvent) (WResult, error)
	// IsActive check whether given uuid+filename is active binlog file, if true return current file offset
	IsActive(uuid, filename string) (bool, int64)
	// Flush flushes the binlog writer.
	Flush() error
}