queue

Documentation

Index

• Constants
• type Queue
  • func NewQueue[T any](client *redis.Client, key string) *Queue[T]
  • func (q *Queue[T]) Dequeue(ctx context.Context) (*QueueItem[T], error)
  • func (q *Queue[T]) DequeueN(ctx context.Context, n int) ([]*QueueItem[T], error)
  • func (q *Queue[T]) Enqueue(ctx context.Context, id string, data T) error
  • func (q *Queue[T]) GetAll(ctx context.Context) ([]*QueueItem[T], error)
  • func (q *Queue[T]) Peek(ctx context.Context) (*QueueItem[T], error)
  • func (q *Queue[T]) Remove(ctx context.Context, id string) error
  • func (q *Queue[T]) Size(ctx context.Context) (int64, error)
  • func (q *Queue[T]) StartProcessor(ctx context.Context, ...)
• type QueueItem

Constants

const POLL_INTERVAL = 5 * time.Second

Default poll interval for processor

const QUEUE_CONCURRENCY = 2

Default concurrency for processor

Types

type Queue

type Queue[T any] struct {
	// contains filtered or unexported fields
}

Prirority queue implementation using redis sorted set

func NewQueue

func NewQueue[T any](client *redis.Client, key string) *Queue[T]

func (*Queue[T]) Dequeue

func (q *Queue[T]) Dequeue(ctx context.Context) (*QueueItem[T], error)

Dequeue removes and returns highest priority item

func (*Queue[T]) DequeueN

func (q *Queue[T]) DequeueN(ctx context.Context, n int) ([]*QueueItem[T], error)

DequeueN removes and returns up to N items

func (*Queue[T]) Enqueue

func (q *Queue[T]) Enqueue(ctx context.Context, id string, data T) error

Add an item to queue ! TODO - now we don't care about priority but we might in the future (e.g. production vs dev environment)

func (*Queue[T]) GetAll

func (q *Queue[T]) GetAll(ctx context.Context) ([]*QueueItem[T], error)

GetAll returns all items in the queue ordered by insertion time

func (*Queue[T]) Peek

func (q *Queue[T]) Peek(ctx context.Context) (*QueueItem[T], error)

Peek returns the highest priority item without removing it

func (*Queue[T]) Remove

func (q *Queue[T]) Remove(ctx context.Context, id string) error

Remove removes an item with the given ID from the queue

func (*Queue[T]) Size

func (q *Queue[T]) Size(ctx context.Context) (int64, error)

Size returns the number of items in the queue

func (*Queue[T]) StartProcessor

func (q *Queue[T]) StartProcessor(ctx context.Context, processor func(ctx context.Context, item *QueueItem[T]) error, jobCounter func(ctx context.Context) (int, error))

type QueueItem

type QueueItem[T any] struct {
	ID         string    `json:"id"`
	Data       T         `json:"data"`
	EnqueuedAt time.Time `json:"enqueued_at"`
	Priority   int       `json:"priority"`
}

Item in queue and metadata