core

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
	//
	Emit(msg *Msg) error
	Close() error
}

type EmptyRouter struct{}

type IFilter interface {
	Configure(configData map[string]interface{}) error
	Filter(msg *Msg) (continueNext bool, err error)
	Close() 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_cache.PositionCacheInterface) error
	Close()
	Stage() config.InputMode
	Done() chan position_repos.Position
	SendDeadSignal() error // for test only
	Wait()
}

type Msg struct {
	Phase

	Type     MsgType
	Host     string
	Database string
	Table    string

	DdlMsg *DDLMsg
	DmlMsg *DMLMsg

	//
	// Timestamp, TimeZone, Oplog will be deprecated.
	//
	Timestamp time.Time // event generated at source
	TimeZone  *time.Location
	Oplog     *gtm.Op

	InputStreamKey *string

	// OutputDepHashed defines the dependency of this msg.
	OutputDepHashes []OutputHash
	Done            chan struct{}

	InputSequence *int64

	InputContext        interface{}
	AfterCommitCallback MsgCallbackFunc
	AfterAckCallback    MsgCallbackFunc
}

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

type MsgCallbackFunc func(m *Msg) error

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

type MsgType string

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

type OutputHash struct {
	Name string
	H    uint64
}

type Phase struct {
	Start          time.Time
	EnterEmitter   time.Time // also leave input
	LeaveEmitter   time.Time
	EnterScheduler time.Time // also enter submitter
	LeaveScheduler time.Time // also leave acker
	LeaveSubmitter time.Time
	EnterAcker     time.Time // also leave output
	EnterOutput    time.Time
}

type PositionCacheCreator interface {
	NewPositionCache() (position_cache.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"`
}