statestorage

Documentation

Index

• func RegisterStateStorage(name config.StateStorageType, storageProvider StorageProvider) bool
• type Manager
  • func NewStateStorageManager(stateStorage Storage) Manager
• type Offset
  • func (o *Offset) Equal(other *Offset) bool
  • func (o *Offset) MarshalBinary() ([]byte, error)
  • func (o *Offset) UnmarshalBinary(data []byte) error
• type StateEncoderFunc
  • func (sef StateEncoderFunc) MarshalBinary() (data []byte, err error)
• type Storage
  • func NewDummyStateStorage() Storage
  • func NewFileStateStorage(path string) (Storage, error)
  • func NewStateStorage(name config.StateStorageType, config *config.Config) (Storage, error)
• type StorageProvider

Functions

func RegisterStateStorage

func RegisterStateStorage(
	name config.StateStorageType, storageProvider StorageProvider,
) bool

RegisterStateStorage registers a config.StateStorageType to a Provider implementation which creates the Storage when requested

Types

type Manager

type Manager interface {
	Start() error
	Stop() error
	Get() (map[string]*Offset, error)
	Set(
		key string, value *Offset,
	) error
	StateEncoder(
		name string, encoder encoding.BinaryMarshaler,
	) error
	StateDecoder(
		name string, decoder encoding.BinaryUnmarshaler,
	) (present bool, err error)
	SetEncodedState(
		name string, encodedState []byte,
	)
	EncodedState(
		name string,
	) (encodedState []byte, present bool)
	SnapshotContext() (*watermark.SnapshotContext, error)
	SnapshotContextTransaction(
		snapshotName string, createIfNotExists bool, transaction func(snapshotContext *watermark.SnapshotContext) error,
	) error
}

func NewStateStorageManager

func NewStateStorageManager(
	stateStorage Storage,
) Manager

type Offset

type Offset struct {
	Timestamp      time.Time   `json:"timestamp"`
	Snapshot       bool        `json:"snapshot"`
	SnapshotName   *string     `json:"snapshot_name,omitempty"`
	SnapshotOffset int         `json:"snapshot_offset"`
	LSN            pgtypes.LSN `json:"lsn"`
}

func (*Offset) Equal

func (o *Offset) Equal(
	other *Offset,
) bool

func (*Offset) MarshalBinary

func (o *Offset) MarshalBinary() ([]byte, error)

func (*Offset) UnmarshalBinary

func (o *Offset) UnmarshalBinary(
	data []byte,
) error

type StateEncoderFunc

type StateEncoderFunc func() (data []byte, err error)

func (StateEncoderFunc) MarshalBinary

func (sef StateEncoderFunc) MarshalBinary() (data []byte, err error)

type Storage

type Storage interface {
	Start() error
	Stop() error
	Save() error
	Load() error
	Get() (map[string]*Offset, error)
	Set(
		key string, value *Offset,
	) error
	StateEncoder(
		name string, encoder encoding.BinaryMarshaler,
	) error
	StateDecoder(
		name string, decoder encoding.BinaryUnmarshaler,
	) (present bool, err error)
	EncodedState(
		name string,
	) (encodedState []byte, present bool)
	SetEncodedState(
		name string, encodedState []byte,
	)
}

func NewDummyStateStorage

func NewDummyStateStorage() Storage

func NewFileStateStorage

func NewFileStateStorage(
	path string,
) (Storage, error)

func NewStateStorage

func NewStateStorage(
	name config.StateStorageType, config *config.Config,
) (Storage, error)

NewStateStorage instantiates a new instance of the requested Storage when available, otherwise returns an error.

type StorageProvider

type StorageProvider = func(config *config.Config) (Storage, error)