queue

Documentation

Index

• Constants
• type ItemHandler
• type Queue
  • func New(ratelimiter workqueue.RateLimiter, name string, handler ItemHandler) *Queue
  • func (q *Queue) Empty() bool
  • func (q *Queue) Enqueue(key string)
  • func (q *Queue) EnqueueAfter(key string, after time.Duration)
  • func (q *Queue) EnqueueWithoutRateLimit(key string)
  • func (q *Queue) Forget(key string)
  • func (q *Queue) Run(ctx context.Context, workers int)

Constants

const (
	// MaxRetries is the number of times we try to process a given key before permanently forgetting it.
	MaxRetries = 20
)

Types

type ItemHandler

type ItemHandler func(ctx context.Context, key string) error

ItemHandler is a callback that handles a single key on the Queue

type Queue

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

Queue implements a wrapper around workqueue with native VK instrumentation

func New

func New(ratelimiter workqueue.RateLimiter, name string, handler ItemHandler) *Queue

New creates a queue

It expects to get a item rate limiter, and a friendly name which is used in logs, and in the internal kubernetes metrics.

func (*Queue) Empty

func (q *Queue) Empty() bool

Empty returns if the queue has no items in it

It should only be used for debugging, as delayed items are not counted, leading to confusion

func (*Queue) Enqueue

func (q *Queue) Enqueue(key string)

Enqueue enqueues the key in a rate limited fashion

func (*Queue) EnqueueAfter

func (q *Queue) EnqueueAfter(key string, after time.Duration)

EnqueueAfter enqueues the item after this period

Since it wrap workqueue semantics, if an item has been enqueued after, and it is immediately scheduled for work, it will process the immediate item, and then upon the latter delayed processing it will be processed again

func (*Queue) EnqueueWithoutRateLimit

func (q *Queue) EnqueueWithoutRateLimit(key string)

EnqueueWithoutRateLimit enqueues the key without a rate limit

func (*Queue) Forget

func (q *Queue) Forget(key string)

Forget forgets the key

func (*Queue) Run

func (q *Queue) Run(ctx context.Context, workers int)

Run starts the workers

It blocks until context is cancelled, and all of the workers exit.