cluster

Documentation

Overview

Package cluster provides a cluster manager. cluster.go contains the Cluster interface and its implementation.

Package cluster provides a cluster manager. mutex.go contains the implementation of Mutex.

Index

• Variables
• type Cluster
  • func New(urls []string, opts ...Option) (Cluster, error)
• type Message
• type Mutex
• type MutexOption
  • func WithDisableKeepalive() MutexOption
• type Option
  • func WithRequestTimeout(timeout time.Duration) Option
• type Queue

Variables

var (
	// ErrInvalidTTL is returned when a TTL is less than minLockTTL.
	ErrInvalidTTL = fmt.Errorf("invalid TTL, must be greater than %s", minLockTTL.String())

	// ErrLocked is returned when a lock locked by another session.
	ErrLocked = errors.New("lock already held by another session")
	// ErrLockedBySelf is returned when a lock locked by self.
	ErrLockedBySelf = errors.New("lock already held by self")
	// ErrLockNotHeld is returned when a lock is not held
	ErrLockNotHeld = errors.New("lock not held")
)

var (
	// ErrKeyExists is returned by putNewKV when the key already exists.
	ErrKeyExists = errors.New("key already exists")
	// ErrNoWatcher is returned when the watcher channel is nil.
	ErrNoWatcher = errors.New("no watcher channel")
)

Types

type Cluster

type Cluster interface {
	// Mutex returns a distributed mutex implementation.
	Mutex(name string, ttl time.Duration, opts ...MutexOption) (Mutex, error)
	// Queue returns a distributed queue implementation.
	Queue(topic string) (Queue, error)

	// WallClock returns the wall clock time
	WallClock() time.Time
}

Cluster is a cluster manager.

func New

func New(urls []string, opts ...Option) (Cluster, error)

New creates a new cluster manager.

type Message

type Message struct {
	// Key is the unique identifier of the message
	Key string
	// Value is the content of the message
	Value string
	// the key is unique in the queue
	Unique bool
}

Message is the unit of the queue

type Mutex

type Mutex interface {
	Lock(ctx context.Context) error
	// Unlock releases the lock.
	// May return ErrLockNotHeld.
	Unlock(ctx context.Context) error

	// IsLocked returns whether the lock is held.
	IsLocked() bool

	// Refresh extends the lock with TTL.
	// recommended use it when keepAlive is false
	// will return ErrLockNotHeld if refresh is unsuccessful.
	Refresh(ctx context.Context) error
}

Mutex is a cluster level mutex.

type MutexOption

type MutexOption func(*mutex)

MutexOption is a function that applies an option to Mutex.

func WithDisableKeepalive

func WithDisableKeepalive() MutexOption

WithDisableKeepalive disables the keepalive feature of Mutex.

type Option

type Option func(*clusterOptions)

Option is a function that applies an option to cluster.

func WithRequestTimeout

func WithRequestTimeout(timeout time.Duration) Option

WithRequestTimeout set the request timeout for etcd requests.

type Queue

type Queue interface {
	// Produce adds an element to the queue.
	Produce(m *Message) (string, error)
	// Consume returns the first element in the queue.
	Consume(ctx context.Context) (*Message, error)
	// Commit removes the element from the queue.
	Commit(msg *Message) error
}

Queue is a distributed queue implementation based on etcd