queue

Documentation

Index

• Variables
• func New(url string) (queue.Broker, error)
• type Acknowledger
  • func (a *Acknowledger) Ack() error
  • func (a *Acknowledger) Reject(requeue bool) error
• type Broker
  • func (b *Broker) Close() error
  • func (b *Broker) Queue(name string) (queue.Queue, error)
• type Configuration
• type JobIter
  • func (i *JobIter) Close() error
  • func (i *JobIter) Next() (*queue.Job, error)
• type Queue
  • func (q *Queue) Consume(advertisedWindow int) (queue.JobIter, error)
  • func (q *Queue) Publish(j *queue.Job) error
  • func (q *Queue) PublishDelayed(j *queue.Job, delay time.Duration) error
  • func (q *Queue) RepublishBuried(conditions ...queue.RepublishConditionFunc) error
  • func (q *Queue) Transaction(txcb queue.TxCallback) error

Variables

var (
	ErrConnectionFailed = errors.NewKind("failed to connect to RabbitMQ: %s")
	ErrOpenChannel      = errors.NewKind("failed to open a channel: %s")
	ErrRetrievingHeader = errors.NewKind("error retrieving '%s' header from message %s")
	ErrRepublishingJobs = errors.NewKind("couldn't republish some jobs : %s")
)

Functions

func New

func New(url string) (queue.Broker, error)

New creates a new AMQPBroker.

Types

type Acknowledger

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

Acknowledger implements the Acknowledger for AMQP.

func (*Acknowledger) Ack

func (a *Acknowledger) Ack() error

Ack signals ackwoledgement.

func (*Acknowledger) Reject

func (a *Acknowledger) Reject(requeue bool) error

Reject signals rejection. If requeue is false, the job will go to the buried queue until Queue.RepublishBuried() is called.

type Broker

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

Broker implements the queue.Broker interface for AMQP, such as RabbitMQ.

func (*Broker) Close

func (b *Broker) Close() error

Close closes all the connections managed by the broker.

func (*Broker) Queue

func (b *Broker) Queue(name string) (queue.Queue, error)

Queue returns the queue with the given name.

type Configuration

type Configuration struct {
	BuriedQueueSuffix        string `envconfig:"BURIED_QUEUE_SUFFIX" default:".buriedQueue"`
	BuriedExchangeSuffix     string `envconfig:"BURIED_EXCHANGE_SUFFIX" default:".buriedExchange"`
	BuriedNonBlockingRetries int    `envconfig:"BURIED_BLOCKING_RETRIES" default:"3"`

	RetriesHeader string `envconfig:"RETRIES_HEADER" default:"x-retries"`
	ErrorHeader   string `envconfig:"ERROR_HEADER" default:"x-error-type"`

	BackoffMin    time.Duration `envconfig:"BACKOFF_MIN" default:"200ms"`
	BackoffMax    time.Duration `envconfig:"BACKOFF_MAX" default:"30s"`
	BackoffFactor float64       `envconfig:"BACKOFF_FACTOR" default:"2"`
}

Configuration AMQP configuration settings, this settings are set using the envinroment varabiles.

var DefaultConfiguration Configuration

DefaultConfiguration contains the default configuration initalized from environment variables.

type JobIter

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

JobIter implements the JobIter interface for AMQP.

func (*JobIter) Close

func (i *JobIter) Close() error

Close closes the channel of the JobIter.

func (*JobIter) Next

func (i *JobIter) Next() (*queue.Job, error)

Next returns the next job in the iter.

type Queue

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

Queue implements the Queue interface for the AMQP.

func (*Queue) Consume

func (q *Queue) Consume(advertisedWindow int) (queue.JobIter, error)

Implements Queue. The advertisedWindow value will be the exact number of undelivered jobs in transit, not just the minium.

func (*Queue) Publish

func (q *Queue) Publish(j *queue.Job) error

Publish publishes the given Job to the Queue.

func (*Queue) PublishDelayed

func (q *Queue) PublishDelayed(j *queue.Job, delay time.Duration) error

PublishDelayed publishes the given Job with a given delay. Delayed messages wont go into the buried queue if they fail.

func (*Queue) RepublishBuried

func (q *Queue) RepublishBuried(conditions ...queue.RepublishConditionFunc) error

RepublishBuried will republish in the main queue those jobs that timed out without Ack or were Rejected with requeue = False and makes comply return true.

func (*Queue) Transaction

func (q *Queue) Transaction(txcb queue.TxCallback) error

Transaction executes the given callback inside a transaction.