core

type AfterMsgCommitFunc func(m *Msg) error

type AsynchronousOutput interface {
	Start(msgAcker MsgAcker) error
	Output
}

type DDLMsg struct {
	Statement string
	AST       ast.StmtNode
}

type DMLMsg struct {
	Operation DMLOp
	Data      map[string]interface{}
	Old       map[string]interface{}
	Pks       map[string]interface{}
}

type DMLOp string

type Emitter interface {
	// Emit use fs to modify messages and submit job to scheduler
	// msg is the message to send
	//
	// TODO better interface for Emit, so that we don't need InputStreamKey...
	//
	Emit(msg *Msg) error
}

type EmptyRouter struct{}

type IFilter interface {
	Configure(configData map[string]interface{}) error
	Filter(msg *Msg) (continueNext bool, err error)
}

type IFilterFactory interface {
	NewFilter() IFilter
}

type IMatcher interface {
	Configure(arg interface{}) error
	Match(msg *Msg) bool
}

type IMatcherFactory interface {
	NewMatcher() IMatcher
}

type IMatcherGroup []IMatcher

type Input interface {
	Start(emitter Emitter, router Router, positionCache position_store.PositionCacheInterface) error
	Close()
	Stage() config.InputMode
	Done() chan position_store.Position
	SendDeadSignal() error // for test only
	Wait()
}

type Metrics struct {
	MsgCreateTime time.Time
	MsgEmitTime   time.Time
	MsgSubmitTime time.Time
	MsgAckTime    time.Time
}

type Msg struct {
	Metrics

	Type     MsgType
	Host     string
	Database string
	Table    string

	DdlMsg *DDLMsg
	DmlMsg *DMLMsg

	//
	// Timestamp, TimeZone, Oplog will be deprecated.
	//
	Timestamp time.Time
	TimeZone  *time.Location
	Oplog     *gtm.Op

	InputStreamKey  *string
	OutputStreamKey *string
	Done            chan struct{}

	InputSequence *int64

	InputContext        interface{}
	AfterCommitCallback AfterMsgCommitFunc
}

type MsgAcker interface {
	AckMsg(msg *Msg) error
}

type MsgSubmitter interface {
	SubmitMsg(msg *Msg) error
}

type MsgType string

type Output interface {
	Execute(msgs []*Msg) error
	GetRouter() Router
	Close()
}

type PositionCacheCreator interface {
	NewPositionCache() (position_store.PositionCacheInterface, error)
}

type Router interface {
	Exists(msg *Msg) bool
}

type Scheduler interface {
	MsgSubmitter
	MsgAcker
	Healthy() bool
	Start(output Output) error
	Close()
}

type SynchronousOutput interface {
	Start() error
	Output
}

type TaskReportStage string

type TaskReportStatus struct {
	Name       string          `json:"name"`
	ConfigHash string          `json:"configHash"`
	Position   string          `json:"position"`
	Stage      TaskReportStage `json:"stage"`
	Version    string          `json:"version"`
}