storeadapter

README

storeadapter

Golang interface for ETCD/ZooKeeper style datastores

storeadapter

The storeadapter is an generalized client for connecting to a Zookeeper/ETCD-like high availability store. Writes are performed concurrently for optimal performance.

fakestoreadapter

Provides a fake in-memory implementation of the storeadapter to allow for unit tests that do not need to spin up a database.

storerunner

Brings up and manages the lifecycle of a live ETCD/ZooKeeper server cluster.

Documentation

Index

• Constants
• func NewError(err error, errType int) *customError
• type Error
• type EventType
• type ExponentialRetryPolicy
  • func (ExponentialRetryPolicy) DelayFor(attempts uint) (time.Duration, bool)
• type RetryPolicy
• type Sleeper
• type StoreAdapter
  • func NewRetryable(storeAdapter StoreAdapter, sleeper Sleeper, retryPolicy RetryPolicy) StoreAdapter
• type StoreNode
  • func (self StoreNode) KeyComponents() []string
  • func (self StoreNode) Lookup(childKey string) (StoreNode, bool)
• type WatchEvent

Constants

const (
	ErrorKeyNotFound = iota
	ErrorNodeIsDirectory
	ErrorNodeIsNotDirectory
	ErrorTimeout
	ErrorInvalidFormat
	ErrorInvalidTTL
	ErrorKeyExists
	ErrorKeyComparisonFailed
	ErrorOther

	ErrorInvalidTTLDescription          = "got an invalid TTL"
	ErrorKeyComparisonFailedDescription = "keys do not match"
	ErrorKeyExistsDescription           = "key already exists in store"
	ErrorKeyNotFoundDescription         = "key does not exist in store"
	ErrorNodeIsDirectoryDescription     = "node is a directory, not a leaf"
	ErrorNodeIsNotDirectoryDescription  = "node is a leaf, not a directory"
)

const (
	InvalidEvent = EventType(iota)
	CreateEvent
	DeleteEvent
	ExpireEvent
	UpdateEvent
)

Functions

func NewError

func NewError(err error, errType int) *customError

Types

type Error

type Error interface {
	error
	Type() int
}

type EventType

type EventType int

type ExponentialRetryPolicy

type ExponentialRetryPolicy struct{}

func (ExponentialRetryPolicy) DelayFor

func (ExponentialRetryPolicy) DelayFor(attempts uint) (time.Duration, bool)

type RetryPolicy

type RetryPolicy interface {
	DelayFor(uint) (time.Duration, bool)
}

type Sleeper

type Sleeper interface {
	Sleep(time.Duration)
}

type StoreAdapter

type StoreAdapter interface {
	// Intiailize connection to server. For a store with no
	// persistent connection, this effectively just tests connectivity.
	Connect() error

	// Create a node and fail if it already exists.
	Create(StoreNode) error

	// Update a node and fail if it does not already exist.
	Update(StoreNode) error

	// CompareAndSwap a node and don't swap if the compare fails.
	CompareAndSwap(oldNode, newNode StoreNode) error
	CompareAndSwapByIndex(prevIndex uint64, newNode StoreNode) error

	// Set multiple nodes at once. If any of them fail,
	// it will return the first error.
	SetMulti(nodes []StoreNode) error

	// Retrieve a node from the store at the given key.
	// Returns an error if it does not exist.
	Get(key string) (StoreNode, error)

	// Recursively get the contents of a key.
	ListRecursively(key string) (StoreNode, error)

	// Delete a set of keys from the store. If any fail to be
	// deleted or don't actually exist, an error is returned.
	Delete(keys ...string) error

	// DeleteLeaves removes a set of empty directories and key-value pairs
	// from the store. If any fail to be deleted or don't actually exist,
	// an error is returned.
	DeleteLeaves(keys ...string) error

	// CompareAndDelete and don't delete if the compare fails.
	CompareAndDelete(...StoreNode) error

	// CompareAndDelete by index and don't delete if the compare fails.
	CompareAndDeleteByIndex(...StoreNode) error

	// Set the ttl on a directory
	UpdateDirTTL(key string, ttl uint64) error

	// Watch a given key recursively for changes. Events will come in on one channel, and watching will stop when a value is sent over the stop channel.
	//
	// Events may be missed, but the watcher will do its best to continue.
	//
	// Returns an error if the watcher cannot initially "attach" to the stream.
	//
	// Otherwise, the caller can assume that the watcher will continue attempting to stream events.
	Watch(key string) (events <-chan WatchEvent, stop chan<- bool, errors <-chan error)

	// Close any live persistent connection, and cleans up any running state.
	Disconnect() error

	// Create a node, keep it there, and send a notification when it is lost. Blocks until the node can be created.
	//
	// To release the node, send a channel value to the releaseNode channel, and read from the channel to ensure it's released.
	//
	// If the store times out, returns an error.
	MaintainNode(storeNode StoreNode) (lostNode <-chan bool, releaseNode chan chan bool, err error)
}

func NewRetryable

func NewRetryable(storeAdapter StoreAdapter, sleeper Sleeper, retryPolicy RetryPolicy) StoreAdapter

type StoreNode

type StoreNode struct {
	Key        string
	Value      []byte
	Dir        bool
	TTL        uint64
	ChildNodes []StoreNode
	Index      uint64
}

func (StoreNode) KeyComponents

func (self StoreNode) KeyComponents() []string

func (StoreNode) Lookup

func (self StoreNode) Lookup(childKey string) (StoreNode, bool)

type WatchEvent

type WatchEvent struct {
	Type     EventType
	Node     *StoreNode
	PrevNode *StoreNode
}