oplog

Documentation

Index

• Variables
• func ApplyMigrations(oplog *OpLog) error
• type OpLog
  • func NewOpLog(store OpStore) (*OpLog, error)
  • func (o *OpLog) Add(ops ...*v1.Operation) error
  • func (o *OpLog) Delete(opID ...int64) error
  • func (o *OpLog) Get(opID int64) (*v1.Operation, error)
  • func (o *OpLog) Query(q Query, f func(*v1.Operation) error) error
  • func (o *OpLog) Subscribe(q Query, f *Subscription)
  • func (o *OpLog) Transform(q Query, f func(*v1.Operation) (*v1.Operation, error)) error
  • func (o *OpLog) Unsubscribe(f *Subscription) error
  • func (o *OpLog) Update(ops ...*v1.Operation) error
• type OpStore
• type OperationEvent
• type Query
  • func (q *Query) Match(op *v1.Operation) bool
• type Subscription

Variables

var (
	ErrStopIteration = errors.New("stop iteration")
	ErrNotExist      = errors.New("operation does not exist")
	ErrExist         = errors.New("operation already exists")

	NullOPID = int64(0)
)

var CurrentVersion = int64(len(migrations))

var SelectAll = Query{}

Functions

func ApplyMigrations

func ApplyMigrations(oplog *OpLog) error

Types

type OpLog

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

func NewOpLog

func NewOpLog(store OpStore) (*OpLog, error)

func (*OpLog) Add

func (o *OpLog) Add(ops ...*v1.Operation) error

func (*OpLog) Delete

func (o *OpLog) Delete(opID ...int64) error

func (*OpLog) Get

func (o *OpLog) Get(opID int64) (*v1.Operation, error)

func (*OpLog) Query

func (o *OpLog) Query(q Query, f func(*v1.Operation) error) error

func (*OpLog) Subscribe

func (o *OpLog) Subscribe(q Query, f *Subscription)

func (*OpLog) Transform

func (o *OpLog) Transform(q Query, f func(*v1.Operation) (*v1.Operation, error)) error

func (*OpLog) Unsubscribe

func (o *OpLog) Unsubscribe(f *Subscription) error

func (*OpLog) Update

func (o *OpLog) Update(ops ...*v1.Operation) error

type OpStore

type OpStore interface {
	Query(q Query, f func(*v1.Operation) error) error
	Get(opID int64) (*v1.Operation, error)
	Add(op ...*v1.Operation) error
	Update(op ...*v1.Operation) error              // returns the previous values of the updated operations OR an error
	Delete(opID ...int64) ([]*v1.Operation, error) // returns the deleted operations OR an error
	Transform(q Query, f func(*v1.Operation) (*v1.Operation, error)) error
	Version() (int64, error)
	SetVersion(version int64) error
}

type OperationEvent

type OperationEvent int

const (
	OPERATION_ADDED OperationEvent = iota
	OPERATION_UPDATED
	OPERATION_DELETED
)

type Query

type Query struct {
	// Filter by fields
	OpIDs      []int64
	PlanID     string
	RepoID     string
	SnapshotID string
	FlowID     int64
	InstanceID string

	// Pagination
	Limit    int
	Offset   int
	Reversed bool
	// contains filtered or unexported fields
}

func (*Query) Match

func (q *Query) Match(op *v1.Operation) bool

type Subscription

type Subscription = func(ops []*v1.Operation, event OperationEvent)