queue

May 26, 2026

Changes in this version

+ var ErrInvalidThresholdOrder = errors.New("queue: maxThreshold must be > minThreshold")

+ var ErrItemDropped = errors.New("item dropped by drop predicate")

+ var ErrNegativeMinThreshold = errors.New("queue: minThreshold must be >= 0")

+ var ErrQueueClosed = errors.New("queue closed")

+ type BackpressureQueue struct

+ func NewBackpressureQueue[T any](capacity int, predicate DropPredicate[T]) *BackpressureQueue[T]

+ func (q *BackpressureQueue[T]) Close()

+ func (q *BackpressureQueue[T]) Dequeue(ctx context.Context) fn.Result[T]

+ func (q *BackpressureQueue[T]) Enqueue(ctx context.Context, item T) error

+ func (q *BackpressureQueue[T]) Len() int

+ func (q *BackpressureQueue[T]) ReceiveChan() <-chan T

+ func (q *BackpressureQueue[T]) TryEnqueue(item T) bool

+ type DropCheckFunc func(queueLen int) bool

+ func RandomEarlyDrop(minThreshold, maxThreshold int, opts ...REDOption) (DropCheckFunc, error)

+ type DropPredicate func(queueLen int, item T) bool

+ func AsDropPredicate[T any](f DropCheckFunc) DropPredicate[T]

+ type REDOption func(*redConfig)

+ func WithRandSource(src func() float64) REDOption

Mar 24, 2020

Changes in this version

+ type CircularBuffer struct

+ func NewCircularBuffer(size int) (*CircularBuffer, error)

+ func (c *CircularBuffer) Add(item interface{})

+ func (c *CircularBuffer) Latest() interface{}

+ func (c *CircularBuffer) List() []interface{}

+ func (c *CircularBuffer) Total() int

Dec 12, 2019

Changes in this version

+ type PriorityQueue struct

+ func (pq *PriorityQueue) Empty() bool

+ func (pq *PriorityQueue) Len() int

+ func (pq *PriorityQueue) Pop() PriorityQueueItem

+ func (pq *PriorityQueue) Push(item PriorityQueueItem)

+ func (pq *PriorityQueue) Top() PriorityQueueItem

+ type PriorityQueueItem interface

+ Less func(other PriorityQueueItem) bool

Feb 7, 2019

Changes in this version

+ type ConcurrentQueue struct

+ func NewConcurrentQueue(bufferSize int) *ConcurrentQueue

+ func (cq *ConcurrentQueue) ChanIn() chan<- interface{}

+ func (cq *ConcurrentQueue) ChanOut() <-chan interface{}

+ func (cq *ConcurrentQueue) Start()

+ func (cq *ConcurrentQueue) Stop()

+ type GCQueue struct

+ func NewGCQueue(newItem func() Recycler, returnQueueSize int, ...) *GCQueue

+ func (q *GCQueue) Return(item Recycler)

+ func (q *GCQueue) Take() Recycler

+ type Recycler interface

+ Recycle func()