pq

Documentation

Index

• Variables
• type PriorityQueue
  • func NewPriorityQueue(b broker.Broker, pri int, prefix string, quietPeriod time.Duration) (*PriorityQueue, error)
  • func (q *PriorityQueue) Close() error
  • func (q *PriorityQueue) Consume(out interface{}) (broker.Consumer, int, error)
  • func (q *PriorityQueue) ConsumeCancel(out interface{}, cancelCh chan struct{}) (broker.Consumer, int, error)
  • func (q *PriorityQueue) Max() int
  • func (q *PriorityQueue) Min() int
  • func (q *PriorityQueue) Publish(payload interface{}, pri int) error

Variables

var (
	// The minimum amount of time to wait after receiving a message for a higher
	// priority message to arrive.
	MinQuietPeriod = 10 * time.Millisecond
)

Types

type PriorityQueue

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

PriorityQueue is a simple wrapper around a relay.Broker to manage a set of queues with varying priority.

func NewPriorityQueue

func NewPriorityQueue(
	b broker.Broker,
	pri int,
	prefix string,
	quietPeriod time.Duration) (*PriorityQueue, error)

NewPriorityQueue returns a new priority queue from which a consumer or producer at a given priority can be easily retrieved.

func (*PriorityQueue) Close

func (q *PriorityQueue) Close() error

Close will call a shutdown on all publishers we have used. By default, all publishers are kept open so that multiple calls to establish the sessions are not always required. This method shuts them all down and resets the pool.

func (*PriorityQueue) Consume

func (q *PriorityQueue) Consume(out interface{}) (broker.Consumer, int, error)

Consume is the public method for consuming data out of a priority queue. It will block until data is received, and returns the priority level of the consumed message along with any errors. The consumer is also returned, which should be Ack'ed and Closed by the caller.

func (*PriorityQueue) ConsumeCancel

func (q *PriorityQueue) ConsumeCancel(
	out interface{}, cancelCh chan struct{}) (broker.Consumer, int, error)

ConsumeCancel allows passing in a channel to signal that we should stop trying to consume a message. Internally this channel will be checked on a short interval, and will shut down the job if the channel has been closed.

func (*PriorityQueue) Max

func (q *PriorityQueue) Max() int

Max returns the highest priority number.

func (*PriorityQueue) Min

func (q *PriorityQueue) Min() int

Min returns the lowest priority number. This is always 0.

func (*PriorityQueue) Publish

func (q *PriorityQueue) Publish(payload interface{}, pri int) error

Publish will publish a message at a given priority. The publisher is automatically closed afterward.