lstore

README

lstore

local store of event streams

Documentation

Index

• Variables
• type AppendResult
• type Blob
• type Config
  • func (conf *Config) IndexedSegmentPath(tailOffset Offset) string
  • func (conf *Config) IndexingSegmentPath() string
  • func (conf *Config) IndexingSegmentTmpPath() string
  • func (conf *Config) RawSegmentPath(tailOffset Offset) string
  • func (conf *Config) TailSegmentPath() string
  • func (conf *Config) TailSegmentTmpPath() string
  • func (conf *Config) TombstoneSegmentPath() string
  • func (conf *Config) TombstoneSegmentTmpPath() string
• type Entry
• type Filter
• type Offset
• type OffsetsCollector
  • func (collector *OffsetsCollector) HandleRow(offset Offset, entry *Entry) error
• type Reader
  • func (reader *Reader) Close() error
  • func (reader *Reader) Refresh() (headMoved bool, tailMoved bool)
  • func (reader *Reader) RefreshTail(ctx context.Context) (tailMoved bool)
  • func (reader *Reader) SearchForward(ctxObj context.Context, req *SearchRequest) error
  • func (reader *Reader) TailOffset() Offset
• type Row
• type RowsCollector
  • func (collector *RowsCollector) HandleRow(offset Offset, entry *Entry) error
• type SearchCallback
• type SearchRequest
• type StopAt
  • func (collector *StopAt) HandleRow(offset Offset, entry *Entry) error
• type Store
  • func New(ctxObj context.Context, cfg *Config) (*Store, error)
  • func (store *Store) Append(ctxObj context.Context, entry *Entry) (Offset, error)
  • func (store *Store) BatchAppend(ctxObj context.Context, resultChan chan<- AppendResult, entries []*Entry)
  • func (store *Store) Config() Config
  • func (store *Store) NewBlobValueFilter(column int, value Blob) Filter
  • func (store *Store) NewReader(ctxObj context.Context) (*Reader, error)
  • func (store *Store) Remove(ctxObj context.Context, removingFrom Offset) error
  • func (store *Store) RotateIndex(ctxObj context.Context) error
  • func (store *Store) Stop(ctx context.Context) error
  • func (store *Store) UpdateIndex(ctxObj context.Context) error

Variables

var SearchAborted = errors.New("search aborted")

SearchAborted should be returned if you want to end the scanForward from callback

var SegmentOverflowError = errors.New("please rotate to new chunk")

Types

type AppendResult

type AppendResult struct {
	Offset Offset
	Error  error
}

type Blob

type Blob string

type Config

type Config struct {
	Directory string
	// contains filtered or unexported fields
}

func (*Config) IndexedSegmentPath

func (conf *Config) IndexedSegmentPath(tailOffset Offset) string

func (*Config) IndexingSegmentPath

func (conf *Config) IndexingSegmentPath() string

func (*Config) IndexingSegmentTmpPath

func (conf *Config) IndexingSegmentTmpPath() string

func (*Config) RawSegmentPath

func (conf *Config) RawSegmentPath(tailOffset Offset) string

func (*Config) TailSegmentPath

func (conf *Config) TailSegmentPath() string

func (*Config) TailSegmentTmpPath

func (conf *Config) TailSegmentTmpPath() string

func (*Config) TombstoneSegmentPath

func (conf *Config) TombstoneSegmentPath() string

func (*Config) TombstoneSegmentTmpPath

func (conf *Config) TombstoneSegmentTmpPath() string

type Entry

type Entry struct {
	IntValues  []int64
	BlobValues []Blob
}

type Filter

type Filter interface {
	// contains filtered or unexported methods
}

type Offset

type Offset uint64

type OffsetsCollector

type OffsetsCollector struct {
	LimitSize int
	Offsets   []Offset
}

func (*OffsetsCollector) HandleRow

func (collector *OffsetsCollector) HandleRow(offset Offset, entry *Entry) error

type Reader

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

Reader is not thread safe, can only be used from one goroutine

func (*Reader) Close

func (reader *Reader) Close() error

func (*Reader) Refresh

func (reader *Reader) Refresh() (headMoved bool, tailMoved bool)

func (*Reader) RefreshTail

func (reader *Reader) RefreshTail(ctx context.Context) (tailMoved bool)

RefreshTail has minimum cost of two cas read, one for store.latestVersion, one for tailSegment.tail

func (*Reader) SearchForward

func (reader *Reader) SearchForward(ctxObj context.Context, req *SearchRequest) error

func (*Reader) TailOffset

func (reader *Reader) TailOffset() Offset

type Row

type Row struct {
	*Entry
	Offset Offset
}

type RowsCollector

type RowsCollector struct {
	LimitSize int
	Rows      []Row
}

func (*RowsCollector) HandleRow

func (collector *RowsCollector) HandleRow(offset Offset, entry *Entry) error

type SearchCallback

type SearchCallback interface {
	HandleRow(offset Offset, entry *Entry) error
}

type SearchRequest

type SearchRequest struct {
	StartOffset Offset
	Filter      Filter
	Callback    SearchCallback
}

type StopAt

type StopAt struct {
	Offset         Offset
	SearchCallback SearchCallback
}

func (*StopAt) HandleRow

func (collector *StopAt) HandleRow(offset Offset, entry *Entry) error

type Store

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

Store is physically a directory, containing multiple files on disk it represents the history by a log of entries

func New

func New(ctxObj context.Context, cfg *Config) (*Store, error)

func (*Store) Append

func (store *Store) Append(ctxObj context.Context, entry *Entry) (Offset, error)

func (*Store) BatchAppend

func (store *Store) BatchAppend(ctxObj context.Context, resultChan chan<- AppendResult, entries []*Entry)

func (*Store) Config

func (store *Store) Config() Config

func (*Store) NewBlobValueFilter

func (store *Store) NewBlobValueFilter(column int, value Blob) Filter

func (*Store) NewReader

func (store *Store) NewReader(ctxObj context.Context) (*Reader, error)

func (*Store) Remove

func (store *Store) Remove(ctxObj context.Context, removingFrom Offset) error

Remove can only remove those rows in indexed segment data still hanging in raw segments can not be removed

func (*Store) RotateIndex

func (store *Store) RotateIndex(ctxObj context.Context) error

func (*Store) Stop

func (store *Store) Stop(ctx context.Context) error

func (*Store) UpdateIndex

func (store *Store) UpdateIndex(ctxObj context.Context) error