client

Documentation

Index

• Variables
• func DefaultRetryPolicy() retry.Policy
• func EnsurePath(client *Client, path string, makeLastNode bool) error
• func RegisterDriver(name string, driver Driver)
• type Backoff
  • func (backoff *Backoff) GetDelay() time.Duration
  • func (backoff *Backoff) Reset()
• type Client
  • func New(driverName, connectionString, basePath string, retryPolicy retry.Policy) (client *Client, err error)
  • func (client *Client) Close()
  • func (client *Client) ConnectionString() string
  • func (client *Client) GetConnection() (Connection, error)
  • func (client *Client) GetCustomConnection(basePath string) (Connection, error)
  • func (client *Client) NewRetryLoop(cancelable func(chan chan error) chan error) retry.Loop
  • func (client *Client) SetRetryPolicy(policy retry.Policy) error
• type Connection
• type Dir
  • func (d *Dir) SetVersion(version interface{})
  • func (d *Dir) Version() interface{}
• type Driver
• type Event
• type EventType
• type Leader
• type Lock
• type Node
• type Transaction

Variables

var (
	// EventNodeCreated is emmited when a node is created
	EventNodeCreated = EventType(1)
	// EventNodeDeleted is emitted when a node is deleted
	EventNodeDeleted = EventType(2)
	// EventNodeDataChanged is emitted when a node's data is changed
	EventNodeDataChanged = EventType(3)
	// EventNodeChildrenChanged is emitted when any of the currently watched node's children is changed
	EventNodeChildrenChanged = EventType(4)
	// EventSession is emitted when a disconnect or reconnect happens
	EventSession = EventType(-1)
	// EventNotWatching is emitted when the current watch expired
	EventNotWatching = EventType(-2)
)

var (
	// ErrEmptyNode is returned when a get is performed and the node does not contain data
	ErrEmptyNode = errors.New("coord-client: empty node")
	// ErrInvalidVersionObj is returned when a node's Version() value is not understood by the underlying driver
	ErrInvalidVersionObj = errors.New("coord-client: invalid version object")
	// ErrInvalidPath is returned when the path to store a node is malformed or illegal
	ErrInvalidPath = errors.New("coord-client: invalid path")
	// ErrSerialization is returned when there is an error serializing a node
	ErrSerialization = errors.New("coord-client: serialization error")
	// ErrInvalidDSN is returned when the given DSN can not use interpreted by driver
	ErrInvalidDSN = errors.New("coord-client: invalid DSN")
	// ErrDriverDoesNotExist is returned when the specified driver is not registered
	ErrDriverDoesNotExist = errors.New("coord-client: driver does not exist")
	// ErrNodeExists is returned when an attempt at creating a node at a path where a node already exists
	ErrNodeExists = errors.New("coord-client: node exists")
	// ErrInvalidRetryPolicy is returned when a nil value is for a client policy
	ErrInvalidRetryPolicy = errors.New("coord-client: invalid retry policy")
	// ErrConnectionNotFound is returned when Close(id) is attemped on a connection id that does not exist
	ErrConnectionNotFound = errors.New("coord-client: connection not found")
	// ErrConnectionClosed is returned when an operation is attemped on a closed connection
	ErrConnectionClosed        = errors.New("coord-client: connection is closed")
	ErrUnknown                 = errors.New("coord-client: unknown error")
	ErrAPIError                = errors.New("coord-client: api error")
	ErrNoNode                  = errors.New("coord-client: node does not exist")
	ErrNoAuth                  = errors.New("coord-client: not authenticated")
	ErrBadVersion              = errors.New("coord-client: version conflict")
	ErrNoChildrenForEphemerals = errors.New("coord-client: ephemeral nodes may not have children")
	ErrNotEmpty                = errors.New("coord-client: node has children")
	ErrSessionExpired          = errors.New("coord-client: session has been expired by the server")
	ErrInvalidACL              = errors.New("coord-client: invalid ACL specified")
	ErrAuthFailed              = errors.New("coord-client: client authentication failed")
	ErrClosing                 = errors.New("coord-client: zookeeper is closing")
	ErrNothing                 = errors.New("coord-client: no server responsees to process")
	ErrSessionMoved            = errors.New("coord-client: session moved to another server, so operation is ignored")
	ErrNoServer                = errors.New("coord-client: could not connect to a server")
)

Functions

func DefaultRetryPolicy

func DefaultRetryPolicy() retry.Policy

DefaultRetryPolicy return a retry.Policy of retry.NTimes(30, time.Millisecond*50)

func EnsurePath

func EnsurePath(client *Client, path string, makeLastNode bool) error

EnsurePath creates the given path with empty nodes if they don't exist; the last node in the path is only created if makeLastNode is true.

func RegisterDriver

func RegisterDriver(name string, driver Driver)

RegisterDriver registers driver under name. This function should only be called in a func init().

Types

type Backoff

type Backoff struct {
	InitialDelay time.Duration // the initial delay
	MaxDelay     time.Duration // The maximum delay
	// contains filtered or unexported fields
}

Backoff controls the exponential backoff used when connection attempts to all zookeepers fail

func (*Backoff) GetDelay

func (backoff *Backoff) GetDelay() time.Duration

GetDelay returns the amount of delay that should be used for the current connection attempt. It will return a randomized value of initialDelay on the first call, and will increase the delay randomly on each subsequent call up to maxDelay. The initial delay and each subsequent delay are randomized to avoid a scenario where multiple instances on the same host all start trying to reconnection. In scenarios like those, we don't want all instances reconnecting in lock-step with each other.

func (*Backoff) Reset

func (backoff *Backoff) Reset()

Reset resets the backoff delay to some random value that is btwn 80-120% of the initialDelay.

We want to randomize the initial delay so in cases where many instances simultaneously
lose all ZK connections, they will not all start trying to reconnect at the same time.

type Client

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

Client is a coordination client that abstracts using services like etcd or zookeeper.

func New

func New(driverName, connectionString, basePath string, retryPolicy retry.Policy) (client *Client, err error)

New returns a client that will create connections using the given driver and connection string. Any retryable operations will use the given retry policy.

func (*Client) Close

func (client *Client) Close()

Close is a shutdown request. It will shutdown all outstanding connections.

func (*Client) ConnectionString

func (client *Client) ConnectionString() string

ConnectionString returns the connection String

func (*Client) GetConnection

func (client *Client) GetConnection() (Connection, error)

GetConnection returns a client.Connection that will be managed by the client. Callers should call close() on thier connections when done. The client will also close connections if close is called on the client.

func (*Client) GetCustomConnection

func (client *Client) GetCustomConnection(basePath string) (Connection, error)

func (*Client) NewRetryLoop

func (client *Client) NewRetryLoop(cancelable func(chan chan error) chan error) retry.Loop

NewRetryLoop returns a retry loop that will call the given cancelable function. If the client.Close() method is called, the retry loop will stop. The cancelable function must be able to accept a 'chan chan error' which will have a value if client.Close() was called. The function must respond to the request or risk making the users of this client non-responsive.

func (*Client) SetRetryPolicy

func (client *Client) SetRetryPolicy(policy retry.Policy) error

SetRetryPolicy sets the given policy on the client

type Connection

type Connection interface {
	Close()
	SetID(int)
	ID() int
	SetOnClose(func(int))
	NewTransaction() Transaction
	Create(path string, node Node) error
	CreateDir(path string) error
	CreateEphemeral(path string, node Node) (string, error)
	CreateIfExists(path string, node Node) error
	CreateEphemeralIfExists(path string, node Node) (string, error)
	Exists(path string) (bool, error)
	ExistsW(path string, done <-chan struct{}) (bool, <-chan Event, error)
	Delete(path string) error
	ChildrenW(path string, done <-chan struct{}) (children []string, ev <-chan Event, err error)
	Children(path string) (children []string, err error)
	Get(path string, node Node) error
	GetW(path string, node Node, done <-chan struct{}) (<-chan Event, error)
	Set(path string, node Node) error
	NewLock(path string) (Lock, error)
	NewLeader(path string) (Leader, error)
}

Connection is the interface that allows interaction with the coordination service

type Dir

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

Dir is an empty path node

func (*Dir) SetVersion

func (d *Dir) SetVersion(version interface{})

SetVersion implements Node

func (*Dir) Version

func (d *Dir) Version() interface{}

Version implements Node

type Driver

type Driver interface {
	GetConnection(dsn, basePath string) (Connection, error)
}

Driver is an interface that allows the coordination.Client to get a connection from a driver

type Event

type Event struct {
	Type EventType
	Path string // For non-session events, the path of the watched node.
	Err  error
}

Event describes a coordination client event

type EventType

type EventType int32

EventType is a numerical type to identify event types.

type Leader

type Leader interface {
	TakeLead(node Node, done <-chan struct{}) (<-chan Event, error)
	ReleaseLead() error
	Current(node Node) error
}

Leader is the interface that a Leaer implementation must implement

type Lock

type Lock interface {
	Lock() error
	Unlock() error
}

Lock is the interface that a lock implemenation must implement

type Node

type Node interface {
	Version() interface{}
	SetVersion(interface{})
}

Node is the interface that a serializable object must implement to be stored in a coordination service

type Transaction

type Transaction interface {
	Create(path string, node Node) Transaction
	Set(path string, node Node) Transaction
	Delete(path string) Transaction
	Commit() error
}