goredis

Documentation

Overview

Package goredis implements a Celery broker using Redis and https://github.com/redis/go-redis.

Index

• Constants
• type Broker
  • func NewBroker(options ...BrokerOption) *Broker
  • func (br *Broker) Observe(queues []string)
  • func (br *Broker) Receive() ([]byte, error)
  • func (br *Broker) Send(m []byte, q string) error
• type BrokerOption
  • func WithClient(c *redis.Client) BrokerOption
  • func WithReceiveTimeout(timeout time.Duration) BrokerOption

Constants

const DefaultReceiveTimeout = 5

DefaultReceiveTimeout defines how many seconds the broker's Receive command should block waiting for results from Redis.

Types

type Broker

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

Broker is a Redis broker that sends/receives messages from specified queues.

func NewBroker

func NewBroker(options ...BrokerOption) *Broker

NewBroker creates a broker backed by Redis. By default, it connects to localhost.

func (*Broker) Observe

func (br *Broker) Observe(queues []string)

Observe sets the queues from which the tasks should be received. Note, the method is not concurrency safe.

func (*Broker) Receive

func (br *Broker) Receive() ([]byte, error)

Receive fetches a Celery task message from a tail of one of the queues in Redis. After a timeout it returns nil, nil.

Celery relies on BRPOP command to process messages fairly, see https://github.com/celery/kombu/issues/166. Redis BRPOP is a blocking list pop primitive. It blocks the connection when there are no elements to pop from any of the given lists. An element is popped from the tail of the first list that is non-empty, with the given keys being checked in the order that they are given, see https://redis.io/commands/brpop/.

Note, the method is not concurrency safe.

func (*Broker) Send

func (br *Broker) Send(m []byte, q string) error

Send inserts the specified message at the head of the queue using LPUSH command. Note, the method is safe to call concurrently.

type BrokerOption

type BrokerOption func(*Broker)

BrokerOption sets up a Broker.

func WithClient

func WithClient(c *redis.Client) BrokerOption

WithClient sets Redis client representing a pool of connections.

func WithReceiveTimeout

func WithReceiveTimeout(timeout time.Duration) BrokerOption

WithReceiveTimeout sets a timeout of how long the broker's Receive command should block waiting for results from Redis. Larger the timeout, longer the client will have to wait for Celery app to exit. Smaller the timeout, more BRPOP commands would have to be sent to Redis.