store

Documentation

Index

• func Register(name string, initFunc InitFunc)
• type DataVersion
• type DistLock
  • func NewDistLock(ctx context.Context, client *clientv3.Client, key string, ...) *DistLock
  • func (dl *DistLock) KeepAliveOnce()
  • func (dl *DistLock) Lock() (err error)
  • func (dl *DistLock) TryLock() (bool, error)
  • func (dl *DistLock) Unlock() error
• type EtcdStore
  • func (store *EtcdStore) Create(ctx context.Context, key string, value []byte) error
  • func (store *EtcdStore) CreateWithTTL(ctx context.Context, key string, value []byte, ttl time.Duration) error
  • func (store *EtcdStore) Delete(ctx context.Context, key string) error
  • func (store *EtcdStore) Get(ctx context.Context, key string) ([]byte, error)
  • func (store *EtcdStore) KeepAlive(ctx context.Context, key string, value []byte, ttl time.Duration) (<-chan *clientv3.LeaseKeepAliveResponse, error)
  • func (store *EtcdStore) NewIDGenerate(ctx context.Context, key string, base int64, timeout time.Duration) (int64, error)
  • func (store *EtcdStore) NewLock(ctx context.Context, key string, timeout time.Duration) *DistLock
  • func (store *EtcdStore) PrefixScan(ctx context.Context, prefix string) ([][]byte, [][]byte, error)
  • func (store *EtcdStore) Put(ctx context.Context, key string, value []byte) error
  • func (store *EtcdStore) PutWithLeaseId(ctx context.Context, key string, value []byte, ttl time.Duration, ...) error
  • func (store *EtcdStore) STM(ctx context.Context, apply func(stm concurrency.STM) error) error
  • func (store *EtcdStore) Update(ctx context.Context, key string, value []byte) error
  • func (store *EtcdStore) WatchPrefix(ctx context.Context, key string) (clientv3.WatchChan, error)
• type InitFunc
• type Store
  • func NewEtcdStore(serverAddrs []string) (Store, error)
  • func OpenStore(implementation string, serverAddress []string) (Store, error)
• type WatcherJob

Functions

func Register

func Register(name string, initFunc InitFunc)

Types

type DataVersion

type DataVersion int64

type DistLock

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

* DistLock is based on vitess lock.go

func NewDistLock

func NewDistLock(ctx context.Context, client *clientv3.Client, key string, timeout time.Duration) *DistLock

func (*DistLock) KeepAliveOnce

func (dl *DistLock) KeepAliveOnce()

func (*DistLock) Lock

func (dl *DistLock) Lock() (err error)

func (*DistLock) TryLock

func (dl *DistLock) TryLock() (bool, error)

func (*DistLock) Unlock

func (dl *DistLock) Unlock() error

type EtcdStore

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

func (*EtcdStore) Create

func (store *EtcdStore) Create(ctx context.Context, key string, value []byte) error

if key already in , it will check version if same insert else ????? if key is not in , it will put

func (*EtcdStore) CreateWithTTL

func (store *EtcdStore) CreateWithTTL(ctx context.Context, key string, value []byte, ttl time.Duration) error

CreateWithTTL will create the key-value if key already in , it will overwrite if key is not in , it will put

func (*EtcdStore) Delete

func (store *EtcdStore) Delete(ctx context.Context, key string) error

func (*EtcdStore) Get

func (store *EtcdStore) Get(ctx context.Context, key string) ([]byte, error)

func (*EtcdStore) KeepAlive

func (store *EtcdStore) KeepAlive(ctx context.Context, key string, value []byte, ttl time.Duration) (<-chan *clientv3.LeaseKeepAliveResponse, error)

func (*EtcdStore) NewIDGenerate

func (store *EtcdStore) NewIDGenerate(ctx context.Context, key string, base int64, timeout time.Duration) (int64, error)

NewIDGenerate create a global uniqueness id

func (*EtcdStore) NewLock

func (store *EtcdStore) NewLock(ctx context.Context, key string, timeout time.Duration) *DistLock

func (*EtcdStore) PrefixScan

func (store *EtcdStore) PrefixScan(ctx context.Context, prefix string) ([][]byte, [][]byte, error)

func (*EtcdStore) Put

func (store *EtcdStore) Put(ctx context.Context, key string, value []byte) error

put kv if already exits it will overwrite

func (*EtcdStore) PutWithLeaseId

func (store *EtcdStore) PutWithLeaseId(ctx context.Context, key string, value []byte, ttl time.Duration, leaseId clientv3.LeaseID) error

func (*EtcdStore) STM

func (store *EtcdStore) STM(ctx context.Context, apply func(stm concurrency.STM) error) error

func (*EtcdStore) Update

func (store *EtcdStore) Update(ctx context.Context, key string, value []byte) error

func (*EtcdStore) WatchPrefix

func (store *EtcdStore) WatchPrefix(ctx context.Context, key string) (clientv3.WatchChan, error)

type InitFunc

type InitFunc func(serverAddr []string) (Store, error)

type Store

type Store interface {
	Put(ctx context.Context, key string, value []byte) error
	Create(ctx context.Context, key string, value []byte) error
	CreateWithTTL(ctx context.Context, key string, value []byte, ttl time.Duration) error
	KeepAlive(ctx context.Context, key string, value []byte, ttl time.Duration) (<-chan *clientv3.LeaseKeepAliveResponse, error)
	PutWithLeaseId(ctx context.Context, key string, value []byte, ttl time.Duration, leaseId clientv3.LeaseID) error
	Update(ctx context.Context, key string, value []byte) error
	Get(ctx context.Context, key string) ([]byte, error)
	PrefixScan(ctx context.Context, prefix string) ([][]byte, [][]byte, error)
	Delete(ctx context.Context, key string) error
	//Here we should not use the STM structure of etcd, but should define a data structure
	//equivalent to STM in the store, and then convert it in etcdstorage.go, on the one hand, the
	//decoupling of etcd and master logic, on the other hand Easy to extend other types of storage
	STM(ctx context.Context, apply func(stm concurrency.STM) error) error
	NewLock(ctx context.Context, key string, timeout time.Duration) *DistLock
	//it to generate increment unique id
	NewIDGenerate(ctx context.Context, key string, base int64, timeout time.Duration) (int64, error)
	WatchPrefix(ctx context.Context, key string) (clientv3.WatchChan, error)
}

func NewEtcdStore

func NewEtcdStore(serverAddrs []string) (Store, error)

NewEtcdStore is used to register etcd store init function

func OpenStore

func OpenStore(implementation string, serverAddress []string) (Store, error)

type WatcherJob

type WatcherJob interface {
	Put(event *clientv3.Event)
	Delete(event *clientv3.Event)
	Start()
	Stop()
}