position_repos

Documentation

Index

• Constants
• func IsPositionStoreEvent(schemaName string, tableName string) bool
• func NewMongoRepoConfig(source *config.MongoConnConfig) *config.GenericPluginConfig
• func NewMySQLRepoConfig(annotation string, source *config.DBConfig) *config.GenericPluginConfig
• type MongoPosition
• type MongoPositionRet
• type Position
  • func (p Position) Validate() error
• type PositionEntity
• type PositionMeta
  • func (meta PositionMeta) Validate() error
• type PositionRepo
  • func NewMemRepo(pipelineName string) PositionRepo
  • func NewMySQLRepo(pipelineName string, annotation string, source *config.DBConfig) PositionRepo
• type PositionValueDecoder
• type PositionValueEncoder
• type PositionWrapper

Constants

const (
	Version       = "1.0"
	MongoRepoName = "mongo-repo"
)

const MemRepoName = "mem-repo"

const MySQLRepoName = "mysql-repo"

Functions

func IsPositionStoreEvent

func IsPositionStoreEvent(schemaName string, tableName string) bool

func NewMongoRepoConfig

func NewMongoRepoConfig(source *config.MongoConnConfig) *config.GenericPluginConfig

func NewMySQLRepoConfig

func NewMySQLRepoConfig(annotation string, source *config.DBConfig) *config.GenericPluginConfig

Types

type MongoPosition

type MongoPosition struct {
	StartPosition   config.MongoPosition `json:"start_position" bson:"start_position"`
	CurrentPosition config.MongoPosition `json:"current_position" bson:"current_position"`
}

MongoPosition and PositionEntity is here to keep backward compatible with previous position format

type MongoPositionRet

type MongoPositionRet struct {
	Version    string `json:"version" bson:"version"`
	Name       string `json:"name" bson:"name"`
	Stage      string `json:"stage" bson:"stage"`
	Value      string `json:"value" bson:"value"`
	LastUpdate string `json:"last_update" bson:"last_update"`
}

MongoPositionRet is the new format

type Position

type Position struct {
	PositionMeta
	Value interface{} `bson:"-" json:"-"`
}

func (Position) Validate

func (p Position) Validate() error

type PositionEntity

type PositionEntity struct {
	Name          string `json:"name" bson:"name"`
	Stage         string `json:"stage" bson:"stage"`
	MongoPosition `json:",inline" bson:",inline"`
	LastUpdate    string `json:"last_update" bson:"last_update"`
}

PositionEntity is the old format, will be deprecated

type PositionMeta

type PositionMeta struct {
	// Version is the schema version of position
	Version string `bson:"version" json:"version"`
	// Name is the unique name of a pipeline
	Name       string
	Stage      config.InputMode
	UpdateTime time.Time
}

func (PositionMeta) Validate

func (meta PositionMeta) Validate() error

type PositionRepo

type PositionRepo interface {
	Init() error
	Get(pipelineName string) (PositionMeta, string, bool, error)
	Put(pipelineName string, positionMeta PositionMeta, v string) error
	Delete(pipelineName string) error
	Close() error
}

func NewMemRepo

func NewMemRepo(pipelineName string) PositionRepo

func NewMySQLRepo

func NewMySQLRepo(pipelineName string, annotation string, source *config.DBConfig) PositionRepo

type PositionValueDecoder

type PositionValueDecoder func(s string) (interface{}, error)

type PositionValueEncoder

type PositionValueEncoder func(v interface{}) (string, error)

type PositionWrapper

type PositionWrapper struct {
	PositionMeta
	MongoValue string `bson:"value" json:"value"`
}