queue

Documentation

Overview

Package queue provides a Redis-backed work queue for the governor.

Index

• Variables
• type Client
  • func NewClient(url string) (*Client, error)
  • func (c *Client) Close() error
  • func (c *Client) Enqueue(ctx context.Context, payload []byte) error
  • func (c *Client) GetDispatchCounter(ctx context.Context, key string) (int64, error)
  • func (c *Client) IncrDispatchCounter(ctx context.Context, key string) (int64, error)
  • func (c *Client) Peek(ctx context.Context) ([]byte, error)
  • func (c *Client) Ping(ctx context.Context) error
• type Queue

Variables

var ErrEmptyQueue = errors.New("queue: empty queue")

ErrEmptyQueue is returned when Peek is called on an empty queue.

var ErrInvalidRedisURL = errors.New("queue: invalid redis URL")

ErrInvalidRedisURL is returned when the Redis URL is empty or cannot be parsed.

var ErrRedisUnavailable = errors.New("queue: redis unavailable")

ErrRedisUnavailable is returned when the Redis server cannot be reached.

Types

type Client

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

Client wraps a *redis.Client and implements the Queue interface.

func NewClient

func NewClient(url string) (*Client, error)

NewClient parses url and returns a connected Client. Returns ErrInvalidRedisURL if url is empty or malformed.

func (*Client) Close

func (c *Client) Close() error

Close releases the underlying Redis connection pool.

func (*Client) Enqueue

func (c *Client) Enqueue(ctx context.Context, payload []byte) error

Enqueue appends payload to the tail of the queue using RPUSH.

func (*Client) GetDispatchCounter

func (c *Client) GetDispatchCounter(ctx context.Context, key string) (int64, error)

GetDispatchCounter returns the current integer value stored at key. Returns 0 if the key does not exist.

func (*Client) IncrDispatchCounter

func (c *Client) IncrDispatchCounter(ctx context.Context, key string) (int64, error)

IncrDispatchCounter atomically increments the named counter key and returns the resulting value.

func (*Client) Peek

func (c *Client) Peek(ctx context.Context) ([]byte, error)

Peek returns the oldest payload (head of the list) without removing it. Returns ErrEmptyQueue when the list is empty.

func (*Client) Ping

func (c *Client) Ping(ctx context.Context) error

Ping verifies connectivity to Redis. Returns ErrRedisUnavailable (wrapping the underlying error) on failure.

type Queue

type Queue interface {
	// Ping verifies connectivity to the backing store.
	Ping(ctx context.Context) error

	// Enqueue appends payload to the tail of the queue.
	Enqueue(ctx context.Context, payload []byte) error

	// Peek returns the oldest payload without removing it.
	// Returns ErrEmptyQueue when the queue is empty.
	Peek(ctx context.Context) ([]byte, error)

	// IncrDispatchCounter atomically increments the named counter and
	// returns the new value.
	IncrDispatchCounter(ctx context.Context, key string) (int64, error)

	// GetDispatchCounter returns the current value of the named counter,
	// or 0 if it has never been set.
	GetDispatchCounter(ctx context.Context, key string) (int64, error)

	// Close releases resources held by the client.
	Close() error
}

Queue defines the interface for the governor's work queue operations. Implementations must be safe for concurrent use.