backend

Documentation

Index

• Variables
• func PrefixEnd(prefix []byte) []byte
• type Backend
  • func NewBackend(kv storage.KvStorage, config Config, metricCli metrics.Metrics) Backend
• type Config
• type FindRet
• type Key
  • func (k Key) String() string
• type KeyVal
• type Ring
  • func NewRing(l int) *Ring
  • func (r *Ring) Add(event *proto.Event)
  • func (r *Ring) FindEvents(revision uint64) (ret *FindRet)
  • func (r *Ring) Reset()
  • func (r *Ring) Size() int
• type WatcherHub
  • func (w *WatcherHub) AddWatcher(ctx context.Context) (<-chan []*proto.Event, error)
  • func (w *WatcherHub) DeleteWatcher(sub chan []*proto.Event, lock bool)
  • func (w *WatcherHub) Stream(input chan []*proto.Event)

Variables

var ErrRevisionDriftBack = errors.New("revision drift back")

Functions

func PrefixEnd

func PrefixEnd(prefix []byte) []byte

Types

type Backend

type Backend interface {

	// Create inserts new key into storage
	Create(ctx context.Context, request *proto.CreateRequest) (*proto.CreateResponse, error)

	// Update set key into storage
	Update(ctx context.Context, request *proto.UpdateRequest) (*proto.UpdateResponse, error)

	// Delete removes key from storage
	Delete(ctx context.Context, request *proto.DeleteRequest) (*proto.DeleteResponse, error)

	// Compact clears the kvs that are too old
	Compact(ctx context.Context, revision uint64) (*proto.CompactResponse, error)

	// Get read a kv from storage
	Get(ctx context.Context, r *proto.GetRequest) (*proto.GetResponse, error)

	// List read kvs in range
	List(ctx context.Context, r *proto.RangeRequest) (*proto.RangeResponse, error)

	// Count counts the number of kvs in range
	Count(ctx context.Context, r *proto.CountRequest) (*proto.CountResponse, error)

	// GetPartitions query the partition state of storage for ListByStream
	GetPartitions(ctx context.Context, r *proto.ListPartitionRequest) (*proto.ListPartitionResponse, error)

	// ListByStream reads kvs in range by stream
	ListByStream(ctx context.Context, startKey, endKey []byte, revision uint64) (<-chan *proto.StreamRangeResponse, error)

	// Watch subscribe the changes from revision on kvs with given prefix
	Watch(ctx context.Context, key string, revision uint64) (<-chan []*proto.Event, error)

	// GetResourceLock returns the resource lock for leader election
	GetResourceLock() resourcelock.Interface

	// GetCurrentRevision returns the read revision
	GetCurrentRevision() uint64

	// SetCurrentRevision is used for init tso for leader
	SetCurrentRevision(uint64)
}

func NewBackend

func NewBackend(kv storage.KvStorage, config Config, metricCli metrics.Metrics) Backend

NewBackend builds a new backend

type Config

type Config struct {
	// EnableEtcdCompatibility make backend compatible with etcd3
	EnableEtcdCompatibility bool

	// Prefix is the range that backend is in charge of
	Prefix string

	// Identity is the identity for a unique backend
	Identity string

	// SkippedPrefixes is the range that backend is not in charge of
	SkippedPrefixes []string

	// WatchCacheSize is the cache size of events
	WatchCacheSize int
}

Config is the configuration for backend

type FindRet

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

type Key

type Key []byte

func (Key) String

func (k Key) String() string

type KeyVal

type KeyVal struct {
	Key      []byte
	Val      []byte
	Revision uint64
}

type Ring

type Ring struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func NewRing

func NewRing(l int) *Ring

func (*Ring) Add

func (r *Ring) Add(event *proto.Event)

func (*Ring) FindEvents

func (r *Ring) FindEvents(revision uint64) (ret *FindRet)

FindEvents return event start from(inclusive) revision

func (*Ring) Reset

func (r *Ring) Reset()

func (*Ring) Size

func (r *Ring) Size() int

type WatcherHub

type WatcherHub struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

WatcherHub maintain registry of Watcher

func (*WatcherHub) AddWatcher

func (w *WatcherHub) AddWatcher(ctx context.Context) (<-chan []*proto.Event, error)

AddWatcher add watcher, filter by prefix and revision is processed in upper server layer

func (*WatcherHub) DeleteWatcher

func (w *WatcherHub) DeleteWatcher(sub chan []*proto.Event, lock bool)

DeleteWatcher delete watcher

func (*WatcherHub) Stream

func (w *WatcherHub) Stream(input chan []*proto.Event)

Stream push events to watchers.