queue

func DefaultConfig ¶

func DefaultConfig() *Config

DefaultConfig returns default configuration

func WithConfig ¶

func WithConfig(cfg *Config) Option

WithConfig sets the queue configuration

func WithProcessor ¶

func WithProcessor(processor TaskProcessor) Option

WithProcessor sets a custom task processor

func NewPriorityQueue ¶

func NewPriorityQueue(capacity int, maxPriority int) *PriorityQueue

NewPriorityQueue creates a new priority queue

Usage:

// Initialize a priority queue with capacity 1000 and maximum priority 10 pq := NewPriorityQueue(1000, 10)

// Add tasks to the queue

task1 := &QueuedTask{
    ID:       "task1",
    Priority: 5,
    Data:     "data1",
}

pq.Push(task1)

task2 := &QueuedTask{
    ID:       "task2",
    Priority: 8,
    Data:     "data2",
}

pq.Push(task2)

// Retrieve the highest priority task highestTask := pq.Pop() // Returns task2

// Update the priority of a specific task pq.UpdatePriority("task1", 9)

// Retrieve a batch of tasks (up to 5) tasks := pq.PopBatch(5)

// Retrieve queue metrics metrics := pq.GetMetrics()

func NewTaskQueue ¶

func NewTaskQueue(ctx context.Context, opts ...Option) *TaskQueue

NewTaskQueue creates a new task queue

Usage:

// Define a custom task
type CustomTask struct {
    Name string
    Data interface{}
}

// Implement TaskProcessor for CustomTask
type CustomTaskProcessor struct{}

func (p *CustomTaskProcessor) Process(task *queue.QueuedTask) error {
    // Convert task data to CustomTask
    customTask, ok := task.Data.(*CustomTask)
    if !ok {
        return fmt.Errorf("invalid task data type")
    }

    // Process the task
    fmt.Printf("Processing task: %s with data: %v", customTask.Name, customTask.Data)
    time.Sleep(time.Second) // Simulate task execution
    return nil
}

// Create queue with default processor (handles func() and func() error)
q1 := NewTaskQueue(context.Background())

// Add a simple task using default processor
err := q1.Push(&QueuedTask{
    ID: "func-task",
    Data: func() error {
        // Do some work
        return nil
    },
})

// Configuration for the task queue
cfg := &Config{
    Workers:     10,
    QueueSize:   1000,
    TaskTimeout: time.Minute,
    MaxPriority: 5,
    MaxRetries:  3,
    RetryDelay:  time.Second * 5,
}

// Create queue with custom processor and configuration
processor := &CustomTaskProcessor{}
q2 := NewTaskQueue(context.Background(),
    WithProcessor(processor),
    WithConfig(cfg),
)

// Start the task queue
q2.Start()

// Example tasks
normalTask := &QueuedTask{
    ID:   "task-1",
    Type: "normal",
    Data: &CustomTask{
        Name: "Normal Task",
        Data: "some data",
    },
}

priorityTask := &QueuedTask{
    ID:       "task-2",
    Type:     "priority",
    Priority: 5,
    Data: &CustomTask{
        Name: "Priority Task",
        Data: "priority data",
    },
}

timerTask := &QueuedTask{
    ID:        "task-3",
    Type:      "timer",
    TriggerAt: time.Now().Add(time.Minute),
    Data: &CustomTask{
        Name: "Timer Task",
        Data: "timer data",
    },
}

// Add tasks to the queue
if err := q2.Push(normalTask); err != nil {
    fmt.Printf("Failed to push normal task: %v", err)
}

if err := q2.Push(priorityTask); err != nil {
    fmt.Printf("Failed to push priority task: %v", err)
}

if err := q2.Push(timerTask); err != nil {
    fmt.Printf("Failed to push timer task: %v", err)
}

// Retrieve queue metrics
metrics := q2.GetMetrics()

// Stop the task queue with a timeout
stopCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
q2.Stop(stopCtx)

func NewTimerQueue ¶

func NewTimerQueue(capacity int) *TimerQueue

NewTimerQueue creates a new timer queue with the specified capacity

Usage:

// Initialize a timer queue with capacity 1000 tq := NewTimerQueue(1000)

// Add a timer task

task := &QueuedTask{
    ID:        "timer-1",
    TriggerAt: time.Now().Add(time.Minute),
    Data:      "timer data",
}

tq.Push(task)

// Retrieve the duration until the next task is due nextDue := tq.NextDue() fmt.Printf("Next task due in: %v\n", nextDue)