changelog

Documentation

Index

• func GetLastIndex(dir string) (index uint64, err error)
• func LogPath(dir string) string
• func MockKVPairs(kvPairs ...string) []*iavl.KVPair
• type Config
• type Message
• type Stream
  • func NewStream(logger logger.Logger, dir string, config Config) (*Stream, error)
  • func (stream *Stream) CheckError() error
  • func (stream *Stream) Close() error
  • func (stream *Stream) FirstOffset() (index uint64, err error)
  • func (stream *Stream) LastOffset() (index uint64, err error)
  • func (stream *Stream) ReadAt(index uint64) (*proto.ChangelogEntry, error)
  • func (stream *Stream) Replay(start uint64, end uint64, ...) error
  • func (stream *Stream) TruncateAfter(index uint64) error
  • func (stream *Stream) TruncateBefore(index uint64) error
  • func (stream *Stream) Write(offset uint64, entry proto.ChangelogEntry) error
• type Subscriber
  • func NewSubscriber(maxPendingSize int, processFn func(entry proto.ChangelogEntry) error) *Subscriber
  • func (s *Subscriber) CheckError() error
  • func (s *Subscriber) Close() error
  • func (s *Subscriber) ProcessEntry(entry proto.ChangelogEntry) error
  • func (s *Subscriber) Start()

Functions

func GetLastIndex

func GetLastIndex(dir string) (index uint64, err error)

GetLastIndex returns the last written index of the replay log

func LogPath

func LogPath(dir string) string

func MockKVPairs

func MockKVPairs(kvPairs ...string) []*iavl.KVPair

Types

type Config

type Config struct {
	DisableFsync    bool
	ZeroCopy        bool
	WriteBufferSize int
}

type Message

type Message struct {
	Index uint64
	Data  *proto.ChangelogEntry
}

type Stream

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

func NewStream

func NewStream(logger logger.Logger, dir string, config Config) (*Stream, error)

NewStream creates a new changelog stream that persist the changesets in the log

func (*Stream) CheckError

func (stream *Stream) CheckError() error

CheckError check if there's any failed async writes or not

func (*Stream) Close

func (stream *Stream) Close() error

func (*Stream) FirstOffset

func (stream *Stream) FirstOffset() (index uint64, err error)

func (*Stream) LastOffset

func (stream *Stream) LastOffset() (index uint64, err error)

LastOffset returns the last written offset/index of the log

func (*Stream) ReadAt

func (stream *Stream) ReadAt(index uint64) (*proto.ChangelogEntry, error)

ReadAt will read the log entry at the provided index

func (*Stream) Replay

func (stream *Stream) Replay(start uint64, end uint64, processFn func(index uint64, entry proto.ChangelogEntry) error) error

Replay will read the replay log and process each log entry with the provided function

func (*Stream) TruncateAfter

func (stream *Stream) TruncateAfter(index uint64) error

TruncateAfter will remove all entries that are after the provided `index`. In other words the entry at `index` becomes the last entry in the log.

func (*Stream) TruncateBefore

func (stream *Stream) TruncateBefore(index uint64) error

TruncateBefore will remove all entries that are before the provided `index`. In other words the entry at `index` becomes the first entry in the log.

func (*Stream) Write

func (stream *Stream) Write(offset uint64, entry proto.ChangelogEntry) error

Write will write a new entry to the log at given index. Whether the writes is in blocking or async manner depends on the buffer size.

type Subscriber

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

func NewSubscriber

func NewSubscriber(
	maxPendingSize int,
	processFn func(entry proto.ChangelogEntry) error,
) *Subscriber

func (*Subscriber) CheckError

func (s *Subscriber) CheckError() error

func (*Subscriber) Close

func (s *Subscriber) Close() error

func (*Subscriber) ProcessEntry

func (s *Subscriber) ProcessEntry(entry proto.ChangelogEntry) error

func (*Subscriber) Start

func (s *Subscriber) Start()