eventstore

package module
v0.0.16 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 15, 2021 License: MIT Imports: 13 Imported by: 1

README

EventStore

GoDoc

High performance event store which is used by Brobridge Gravity.

Benchmark

Here is benchmark results:

$ go test -v -bench Benchmark -run Benchmark
goos: darwin
goarch: amd64
pkg: github.com/BrobridgeOrg/EventStore
BenchmarkWrite
BenchmarkWrite-16                  	  514297	      2336 ns/op
BenchmarkEventThroughput
BenchmarkEventThroughput-16        	  223045	      5077 ns/op
BenchmarkSnapshotPerformance
BenchmarkSnapshotPerformance-16    	  108236	     11125 ns/op
PASS
ok  	github.com/BrobridgeOrg/EventStore	15.152s

License

Licensed under the MIT License

Authors

Copyright(c) 2020 Fred Chien fred@brobridge.com

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func BytesToUint64 

func BytesToUint64(data []byte) uint64

func GetFixedSizeBytes 

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

func StrToBytes 

func StrToBytes(s string) []byte

func Uint64ToBytes 

func Uint64ToBytes(n uint64) []byte

Types

type ColumnFamily added in v0.0.14 

type ColumnFamily struct {
	Store *Store
	Db    *pebble.DB
	Name  string
	Merge func([]byte, []byte) []byte
	// contains filtered or unexported fields
}

func NewColumnFamily added in v0.0.14 

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

func (*ColumnFamily) Close added in v0.0.14 

func (cf *ColumnFamily) Close() error

func (*ColumnFamily) Open added in v0.0.14 

func (cf *ColumnFamily) Open() error

func (*ColumnFamily) Write added in v0.0.16 

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

type Counter 

type Counter uint64

func (*Counter) Count 

func (counter *Counter) Count() uint64

func (*Counter) Increase 

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

func (*Counter) SetCount 

func (counter *Counter) SetCount(val uint64)

func (*Counter) String 

func (counter *Counter) String() string

type Event 

type Event struct {
	Sequence uint64
	Data     []byte
	// contains filtered or unexported fields
}

func NewEvent 

func NewEvent() *Event

func (*Event) Ack 

func (event *Event) Ack()

type EventStore 

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

func CreateEventStore 

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

func (*EventStore) Close 

func (eventstore *EventStore) Close()

func (*EventStore) GetStore 

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

func (*EventStore) RecoverSnapshot added in v0.0.2 

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

func (*EventStore) SetSnapshotHandler 

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

func (*EventStore) TakeSnapshot 

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

func (*EventStore) UnregisterStore 

func (eventstore *EventStore) UnregisterStore(name string)

type Merger added in v0.0.15 

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

func (*Merger) Finish added in v0.0.15 

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

func (*Merger) MergeNewer added in v0.0.15 

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

func (*Merger) MergeOlder added in v0.0.15 

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

func (*Merger) SetHandler added in v0.0.15 

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

type Options 

type Options struct {
	DatabasePath    string
	EnabledSnapshot bool
	SnapshotOptions *SnapshotOptions
}

func NewOptions 

func NewOptions() *Options

type Record added in v0.0.12 

type Record struct {
	Key  []byte
	Data []byte
}

func NewRecord added in v0.0.12 

func NewRecord() *Record

type SnapshotController 

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

func NewSnapshotController 

func NewSnapshotController(options *SnapshotOptions) *SnapshotController

func (*SnapshotController) RecoverSnapshot added in v0.0.2 

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

func (*SnapshotController) Request 

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

func (*SnapshotController) SetHandler 

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

type SnapshotOptions 

type SnapshotOptions struct {
	WorkerCount int32
	BufferSize  int
}

func NewSnapshotOptions 

func NewSnapshotOptions() *SnapshotOptions

type SnapshotRequest 

type SnapshotRequest struct {
	Store    *Store
	Sequence uint64
	Data     []byte
}

func NewSnapshotRequest 

func NewSnapshotRequest() *SnapshotRequest

func (*SnapshotRequest) Delete 

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

func (*SnapshotRequest) Get 

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

func (*SnapshotRequest) UpdateDurableState added in v0.0.2 

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

func (*SnapshotRequest) Upsert 

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

type SnapshotView added in v0.0.8 

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

func NewSnapshotView added in v0.0.8 

func NewSnapshotView(store *Store) *SnapshotView

func (*SnapshotView) Fetch added in v0.0.8 

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

func (*SnapshotView) Initialize added in v0.0.8 

func (sv *SnapshotView) Initialize() error

func (*SnapshotView) Release added in v0.0.8 

func (sv *SnapshotView) Release() error

type Store 

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

func NewStore 

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

func (*Store) Close 

func (store *Store) Close()

func (*Store) CreateSnapshotView added in v0.0.8 

func (store *Store) CreateSnapshotView() *SnapshotView

func (*Store) DispatchEvent 

func (store *Store) DispatchEvent()

func (*Store) Fetch added in v0.0.5 

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

func (*Store) GetColumnFamailyHandle 

func (store *Store) GetColumnFamailyHandle(name string) (*ColumnFamily, error)

func (*Store) GetDurableState 

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

func (*Store) GetLastSequence 

func (store *Store) GetLastSequence() uint64

func (*Store) Subscribe 

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

func (*Store) UpdateDurableState 

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

func (*Store) Write 

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

type StoreHandler 

type StoreHandler func(*Event)

type Subscription 

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

func NewSubscription 

func NewSubscription(store *Store, durableName string, startAt uint64, cf *ColumnFamily, fn StoreHandler) *Subscription

func (*Subscription) Close 

func (sub *Subscription) Close()

func (*Subscription) Trigger 

func (sub *Subscription) Trigger() error

func (*Subscription) Watch 

func (sub *Subscription) Watch()

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.