eventstore

Sep 14, 2022

Changes in this version

+ var StateClassStore = []byte("s")

+ var StateGroupEvent = []byte("e")

+ var StateGroupSnapshot = []byte("s")

+ var StatePathEventCount = bytes.Join([][]byte{ ... }, []byte("."))

+ var StatePathEventLastSeq = bytes.Join([][]byte{ ... }, []byte("."))

+ var StatePathEventRev = bytes.Join([][]byte{ ... }, []byte("."))

+ var StatePathSnapshotCount = bytes.Join([][]byte{ ... }, []byte("."))

+ var StatePathSnapshotLastSeq = bytes.Join([][]byte{ ... }, []byte("."))

type Store

+ func (store *Store) SetStateBytesByPath(b *pebble.Batch, key []byte, value []byte) error

+ func (store *Store) SetStateUint64ByPath(b *pebble.Batch, key []byte, value uint64) error

type StoreState

+ func (ss *StoreState) Rev() uint64

Sep 12, 2022

Changes in this version

+ var ErrInvalidKey = errors.New("EventStore: invalid key")

+ var ErrSnapshotNotEnabled = errors.New("EventStore: snapshot not enabled")

+ var ErrStateEntryNotFound = errors.New("store: state entry not found")

+ var PrefixEvent = []byte("e")

+ var PrefixSnapshotData = []byte("d")

+ var PrefixState = []byte("s")

+ func BytesToFloat64(data []byte) float64

+ func BytesToInt64(data []byte) int64

+ func Float64ToBytes(n float64) []byte

+ func Int64ToBytes(n int64) []byte

type ColumnFamily

+ Prefix []byte

+ Symbol []byte

+ func (cf *ColumnFamily) Get(key []byte) ([]byte, io.Closer, error)

+ func (cf *ColumnFamily) List(rawPrefix []byte, targetPrimaryKey []byte, opts *ListOptions) (*Cursor, error)

+ type Cursor struct

+ func (cur *Cursor) Close() error

+ func (cur *Cursor) EOF() bool

+ func (cur *Cursor) GetData() []byte

+ func (cur *Cursor) GetKey() []byte

+ func (cur *Cursor) Next() bool

+ type ListOptions struct

+ Prefix []byte

+ WithoutRawPrefix bool

type SnapshotRequest

+ Batch *pebble.Batch

type Store

+ func (store *Store) DeleteState(b *pebble.Batch, class []byte, group []byte, key []byte) error

+ func (store *Store) Get(seq uint64) ([]byte, error)

+ func (store *Store) GetStateBytes(class []byte, group []byte, key []byte) ([]byte, error)

+ func (store *Store) GetStateFloat64(class []byte, group []byte, key []byte) (float64, error)

+ func (store *Store) GetStateInt64(class []byte, group []byte, key []byte) (int64, error)

+ func (store *Store) GetStateString(class []byte, group []byte, key []byte) (string, error)

+ func (store *Store) GetStateUint64(class []byte, group []byte, key []byte) (uint64, error)

+ func (store *Store) ListStates(class []byte, group []byte, key []byte) (*Cursor, error)

+ func (store *Store) SetStateBytes(b *pebble.Batch, class []byte, group []byte, key []byte, value []byte) error

+ func (store *Store) SetStateFloat64(b *pebble.Batch, class []byte, group []byte, key []byte, value float64) error

+ func (store *Store) SetStateInt64(b *pebble.Batch, class []byte, group []byte, key []byte, value int64) error

+ func (store *Store) SetStateString(b *pebble.Batch, class []byte, group []byte, key []byte, value string) error

+ func (store *Store) SetStateUint64(b *pebble.Batch, class []byte, group []byte, key []byte, value uint64) error

+ func (store *Store) State() *StoreState

+ type StoreOpt func(s *Store)

+ func WithMaxSyncInterval(interval time.Duration) StoreOpt

+ func WithSnapshot(enabled bool) StoreOpt

+ type StoreState struct

+ func (ss *StoreState) Count() uint64

+ func (ss *StoreState) LastSeq() uint64

+ func (ss *StoreState) SnapshotCount() uint64

+ func (ss *StoreState) SnapshotLastSeq() uint64

Sep 8, 2022

Changes in this version

+ func BytesToString(b []byte) string

Sep 4, 2022

Changes in this version

type Store

+ func (store *Store) GetSnapshotLastSequence() uint64

Aug 30, 2022

Changes in this version

type Merger

+ func NewMerger() *Merger

Aug 28, 2022

Changes in this version

type ColumnFamily

+ func (cf *ColumnFamily) UnregisterMerger(key []byte)

type SnapshotView

+ func (sv *SnapshotView) Get(collection []byte, key []byte) ([]byte, error)

Aug 16, 2022

Changes in this version

+ var ErrRecordNotFound = errors.New("EventStore: record not found")

+ type SubOpt func(s *Subscription)

+ func DurableName(durable string) SubOpt

+ func StartAtSequence(seq uint64) SubOpt

Aug 12, 2022

Changes in this version

type Options

+ BypassEventStore bool

Feb 13, 2022

Changes in this version

type ColumnFamily

+ func (cf *ColumnFamily) Delete(key []byte) error

type Store

+ func (store *Store) Delete(seq uint64) error

Jan 17, 2022

Changes in this version

type Event

+ func (event *Event) Release()

type Record

+ func (r *Record) Release()

Aug 10, 2021

Changes in this version

type ColumnFamily

+ func (cf *ColumnFamily) RegisterMerger(key []byte, fn func([]byte, []byte) []byte)

Jun 15, 2021

Changes in this version

type ColumnFamily

+ func (cf *ColumnFamily) Write(key []byte, data []byte) error

Jun 10, 2021

Changes in this version

type ColumnFamily

+ Merge func([]byte, []byte) []byte

+ type Merger struct

+ func (m *Merger) Finish(includesBase bool) ([]byte, io.Closer, error)

+ func (m *Merger) MergeNewer(value []byte) error

+ func (m *Merger) MergeOlder(value []byte) error

+ func (m *Merger) SetHandler(fn func(origin []byte, newValue []byte) []byte)

Jun 8, 2021

Changes in this version

+ type ColumnFamily struct

+ Db *pebble.DB

+ Name string

+ Store *Store

+ func NewColumnFamily(store *Store, name string) *ColumnFamily

+ func (cf *ColumnFamily) Close() error

+ func (cf *ColumnFamily) Open() error

May 29, 2021

Changes in this version

+ type Record struct

+ Data []byte

+ Key []byte

+ func NewRecord() *Record

May 23, 2021

Changes in this version

+ type SnapshotView struct

+ func NewSnapshotView(store *Store) *SnapshotView

+ func (sv *SnapshotView) Fetch(collection []byte, key []byte, offset uint64, count int) ([]*Event, error)

+ func (sv *SnapshotView) Initialize() error

+ func (sv *SnapshotView) Release() error

type Store

+ func (store *Store) CreateSnapshotView() *SnapshotView

May 5, 2021

Changes in this version

type Store

+ func (store *Store) Fetch(startAt uint64, offset uint64, count int) ([]*Event, error)

Dec 11, 2020

Changes in this version

type EventStore

+ func (eventstore *EventStore) RecoverSnapshot(store *Store) error

type SnapshotController

+ func (ss *SnapshotController) RecoverSnapshot(store *Store) error

type SnapshotRequest

+ func (request *SnapshotRequest) UpdateDurableState(collection []byte) error

Dec 8, 2020

Changes in this version

+ func BytesToUint64(data []byte) uint64

+ func GetFixedSizeBytes(size int, str string) []byte

+ func StrToBytes(s string) []byte

+ func Uint64ToBytes(n uint64) []byte

+ type Counter uint64

+ func (counter *Counter) Count() uint64

+ func (counter *Counter) Increase(delta uint64) uint64

+ func (counter *Counter) SetCount(val uint64)

+ func (counter *Counter) String() string

+ type Event struct

+ Data []byte

+ Sequence uint64

+ func NewEvent() *Event

+ func (event *Event) Ack()

+ type EventStore struct

+ func CreateEventStore(options *Options) (*EventStore, error)

+ func (eventstore *EventStore) Close()

+ func (eventstore *EventStore) GetStore(storeName string) (*Store, error)

+ func (eventstore *EventStore) SetSnapshotHandler(fn func(*SnapshotRequest) error)

+ func (eventstore *EventStore) TakeSnapshot(store *Store, seq uint64, data []byte) error

+ func (eventstore *EventStore) UnregisterStore(name string)

+ type Options struct

+ DatabasePath string

+ EnabledSnapshot bool

+ RocksdbOptions *gorocksdb.Options

+ SnapshotOptions *SnapshotOptions

+ func NewOptions() *Options

+ type SnapshotController struct

+ func NewSnapshotController(options *SnapshotOptions) *SnapshotController

+ func (ss *SnapshotController) Request(store *Store, seq uint64, data []byte) error

+ func (ss *SnapshotController) SetHandler(fn func(*SnapshotRequest) error)

+ type SnapshotOptions struct

+ BufferSize int

+ WorkerCount int32

+ func NewSnapshotOptions() *SnapshotOptions

+ type SnapshotRequest struct

+ Data []byte

+ Sequence uint64

+ Store *Store

+ func NewSnapshotRequest() *SnapshotRequest

+ func (request *SnapshotRequest) Delete(collection []byte, key []byte) error

+ func (request *SnapshotRequest) Get(collection []byte, key []byte) ([]byte, error)

+ func (request *SnapshotRequest) Upsert(collection []byte, key []byte, fn func(origin []byte) ([]byte, error)) error

+ type Store struct

+ func NewStore(eventstore *EventStore, storeName string) (*Store, error)

+ func (store *Store) Close()

+ func (store *Store) DispatchEvent()

+ func (store *Store) GetColumnFamailyHandle(name string) (*gorocksdb.ColumnFamilyHandle, error)

+ func (store *Store) GetDurableState(durableName string) (uint64, error)

+ func (store *Store) GetLastSequence() uint64

+ func (store *Store) Subscribe(durableName string, startAt uint64, fn StoreHandler) (*Subscription, error)

+ func (store *Store) UpdateDurableState(durableName string, lastSeq uint64) error

+ func (store *Store) Write(data []byte) (uint64, error)

+ type StoreHandler func(*Event)

+ type Subscription struct

+ func NewSubscription(store *Store, durableName string, startAt uint64, iterator *gorocksdb.Iterator, ...) *Subscription

+ func (sub *Subscription) Close()

+ func (sub *Subscription) Trigger() error

+ func (sub *Subscription) Watch()