queue

Documentation

Overview

Package queue provides a bounded priority queue for event persistence. This implements the backpressure mechanism from REVIEW.md.

Index

• Constants
• type Queue
  • func NewQueue(maxSize int) *Queue
  • func (q *Queue) Close()
  • func (q *Queue) FillPercent() float64
  • func (q *Queue) Len() int
  • func (q *Queue) NotifyCh() <-chan struct{}
  • func (q *Queue) Pop() *QueueItem
  • func (q *Queue) PopBatch(n int) []*QueueItem
  • func (q *Queue) Push(item *QueueItem) (dropped bool)
  • func (q *Queue) Stats() Stats
  • func (q *Queue) Wait(ctx context.Context) bool
• type QueueItem
• type Stats

Constants

const (
	PriorityHigh   = "high"   // message_start, message_stop, usage
	PriorityMedium = "medium" // content_block_start/stop
	PriorityLow    = "low"    // content_block_delta (drop first)
)

Priority levels for events.

Types

type Queue

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

Queue is a bounded priority queue with backpressure support.

func NewQueue

func NewQueue(maxSize int) *Queue

NewQueue creates a new bounded priority queue.

func (*Queue) Close

func (q *Queue) Close()

Close closes the queue.

func (*Queue) FillPercent

func (q *Queue) FillPercent() float64

FillPercent returns the current fill percentage (0-100).

func (*Queue) Len

func (q *Queue) Len() int

Len returns the current queue size.

func (*Queue) NotifyCh

func (q *Queue) NotifyCh() <-chan struct{}

NotifyCh returns a channel that receives notifications when items are added.

func (*Queue) Pop

func (q *Queue) Pop() *QueueItem

Pop removes and returns the highest priority item. Returns nil if the queue is empty.

func (*Queue) PopBatch

func (q *Queue) PopBatch(n int) []*QueueItem

PopBatch removes and returns up to n items.

func (*Queue) Push

func (q *Queue) Push(item *QueueItem) (dropped bool)

Push adds an item to the queue. If the queue is full, it applies backpressure rules: - Drop LOW priority items first - Only drop HIGH priority items as last resort

func (*Queue) Stats

func (q *Queue) Stats() Stats

Stats returns queue statistics.

func (*Queue) Wait

func (q *Queue) Wait(ctx context.Context) bool

Wait blocks until the context is cancelled or the queue is closed. Used by consumers to wait for items.

type QueueItem

type QueueItem struct {
	Data      interface{}
	Priority  string
	FlowID    string
	EventType string
	Timestamp time.Time
	// contains filtered or unexported fields
}

QueueItem represents an item in the priority queue.

type Stats

type Stats struct {
	Size          int
	HighCount     int
	MediumCount   int
	LowCount      int
	DropsTotal    uint64
	DropsLow      uint64
	DropsHigh     uint64
	DropsCritical uint64
}

Stats holds queue statistics.