position_store

Documentation

Index

• Constants
• Variables
• func IsPositionStoreEvent(schemaName string, tableName string) bool
• type MongoPosition
• type Position
  • func (p Position) Validate() error
• type PositionCacheInterface
  • func NewPositionCache(pipelineName string, repo PositionRepo, flushDuration time.Duration) (PositionCacheInterface, error)
• type PositionEntity
• type PositionRepo
  • func NewMemoRepo() PositionRepo
  • func NewMongoPositionRepo(session *mgo.Session) (PositionRepo, error)
  • func NewMySQLRepo(dbConfig *utils.DBConfig, annotation string) (PositionRepo, error)

Constants

const (
	Version = "1.0"
)

Variables

var DefaultFlushPeriod = 5 * time.Second

Functions

func IsPositionStoreEvent

func IsPositionStoreEvent(schemaName string, tableName string) bool

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 Position

type Position struct {
	// Version is the schema version of position
	Version string
	// Name is the unique name of a pipeline
	Name       string
	Stage      config.InputMode
	Value      string
	UpdateTime time.Time
}

func (Position) Validate

func (p Position) Validate() error

type PositionCacheInterface

type PositionCacheInterface interface {
	Start() error
	Close()
	Put(position Position) error
	Get() (position Position, exist bool, err error)
	Flush() error
	Clear() error
}

func NewPositionCache

func NewPositionCache(pipelineName string, repo PositionRepo, flushDuration time.Duration) (PositionCacheInterface, 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"`
}

type PositionRepo

type PositionRepo interface {
	Get(pipelineName string) (Position, bool, error)
	Put(pipelineName string, position Position) error
	Delete(pipelineName string) error
	Close() error
}

func NewMemoRepo

func NewMemoRepo() PositionRepo

func NewMongoPositionRepo

func NewMongoPositionRepo(session *mgo.Session) (PositionRepo, error)

func NewMySQLRepo

func NewMySQLRepo(dbConfig *utils.DBConfig, annotation string) (PositionRepo, error)