lightqueue

LightQueue

A lightweight, high-performance message queue library for Go with SQLite persistence, priority handling, and built-in reliability features.

Features

• 🚀 High Performance: In-memory priority heap with batch SQLite persistence
• 🔒 Reliable: Message acknowledgment, retry logic, and dead letter queue
• ⚡ Lightweight: No external dependencies beyond SQLite
• 🎯 Priority Support: Process high-priority messages first
• ⏰ TTL & Cleanup: Automatic message expiration and cleanup
• 📊 Monitoring: Built-in statistics and health monitoring
• 🔄 Fault Tolerant: Graceful shutdown and message recovery

Installation

go get github.com/SachinHg/lightqueue

Quick Start

package main

import (
    "fmt"
    "log"
    "time"
    
    "github.com/SachinHg/lightqueue"
)

func main() {
    // Create queue configuration
    config := &lightqueue.QueueConfig{
        Name:           "my_queue",
        DataDir:        "./queue_data",
        BufferSize:     100,
        FlushInterval:  1 * time.Second,
        MaxRetries:     3,
        MessageTTL:     1 * time.Hour,
        EnableMetrics:  true,
    }
    
    // Create and start queue
    queue, err := lightqueue.NewQueue(config)
    if err != nil {
        log.Fatal(err)
    }
    defer queue.Stop()
    
    // Publish a message
    err = queue.Publish("notifications", "Hello, World!", 1)
    if err != nil {
        log.Fatal(err)
    }
    
    // Consume and process message
    message, err := queue.Consume("notifications", "worker-1")
    if err != nil {
        log.Fatal(err)
    }
    
    fmt.Printf("Received: %v\n", message.Data)
    
    // Acknowledge successful processing
    queue.Acknowledge(message.ID)
}

Real-World Use Cases

1. 🎮 Game Event Processing

Perfect for handling game events, achievements, and player notifications with priority levels.

type GameEvent struct {
    PlayerID string `json:"player_id"`
    Event    string `json:"event"`
    Data     map[string]interface{} `json:"data"`
}

func setupGameEventQueue() *lightqueue.Queue {
    config := &lightqueue.QueueConfig{
        Name:           "game_events",
        DataDir:        "./game_data",
        BufferSize:     500,
        FlushInterval:  2 * time.Second,
        MaxRetries:     2,
        MessageTTL:     30 * time.Minute,
    }
    
    queue, _ := lightqueue.NewQueue(config)
    return queue
}

func publishGameEvent(queue *lightqueue.Queue, event GameEvent, priority int) {
    // Priority: 5=Critical (player death), 3=Important (level up), 1=Normal (item pickup)
    queue.Publish("game_events", event, priority)
}

func processGameEvents(queue *lightqueue.Queue) {
    for {
        message, err := queue.Consume("game_events", "game-processor")
        if err != nil {
            time.Sleep(100 * time.Millisecond)
            continue
        }
        
        event := message.Data.(GameEvent)
        if processEvent(event) {
            queue.Acknowledge(message.ID)
        } else {
            queue.Reject(message.ID, "processing failed")
        }
    }
}

2. 📧 Email Service with Priority

Handle email notifications with different priority levels and retry logic.

type EmailJob struct {
    To       string `json:"to"`
    Subject  string `json:"subject"`
    Body     string `json:"body"`
    Template string `json:"template"`
}

func setupEmailQueue() *lightqueue.Queue {
    config := &lightqueue.QueueConfig{
        Name:           "email_service",
        DataDir:        "./email_data",
        BufferSize:     200,
        FlushInterval:  5 * time.Second,
        MaxRetries:     5, // Important for email delivery
        MessageTTL:     24 * time.Hour,
    }
    
    queue, _ := lightqueue.NewQueue(config)
    return queue
}

func sendEmails() {
    queue := setupEmailQueue()
    defer queue.Stop()
    
    // High priority: Password reset
    queue.Publish("emails", EmailJob{
        To: "[email protected]", 
        Subject: "Password Reset",
        Template: "password_reset",
    }, 10)
    
    // Medium priority: Order confirmation
    queue.Publish("emails", EmailJob{
        To: "[email protected]",
        Subject: "Order Confirmation",
        Template: "order_confirm",
    }, 5)
    
    // Low priority: Newsletter
    queue.Publish("emails", EmailJob{
        To: "[email protected]",
        Subject: "Weekly Newsletter",
        Template: "newsletter",
    }, 1)
    
    // Process emails
    go func() {
        for {
            message, err := queue.Consume("emails", "email-worker")
            if err != nil {
                time.Sleep(1 * time.Second)
                continue
            }
            
            job := message.Data.(EmailJob)
            if sendEmail(job) {
                queue.Acknowledge(message.ID)
            } else {
                queue.Reject(message.ID, "SMTP error")
            }
        }
    }()
}

3. 🖼️ Image Processing Pipeline

Process user uploads with different priorities and handle large workloads.

type ImageTask struct {
    ImageID    string   `json:"image_id"`
    UserID     string   `json:"user_id"`
    Operations []string `json:"operations"` // resize, compress, watermark
    Sizes      []string `json:"sizes"`      // thumbnail, medium, large
}

func setupImageProcessing() {
    queue, _ := lightqueue.NewQueue(&lightqueue.QueueConfig{
        Name:           "image_processor",
        DataDir:        "./image_data",
        BufferSize:     1000,
        FlushInterval:  3 * time.Second,
        MaxRetries:     3,
        MessageTTL:     2 * time.Hour,
    })
    
    // Premium user uploads (high priority)
    queue.Publish("images", ImageTask{
        ImageID: "img_123",
        UserID:  "premium_user",
        Operations: []string{"resize", "compress", "watermark"},
        Sizes: []string{"thumbnail", "medium", "large"},
    }, 8)
    
    // Regular user uploads (normal priority)
    queue.Publish("images", ImageTask{
        ImageID: "img_456", 
        UserID:  "regular_user",
        Operations: []string{"resize", "compress"},
        Sizes: []string{"thumbnail", "medium"},
    }, 3)
    
    // Start multiple workers
    for i := 0; i < 4; i++ {
        go imageWorker(queue, fmt.Sprintf("worker-%d", i))
    }
}

func imageWorker(queue *lightqueue.Queue, workerID string) {
    for {
        message, err := queue.Consume("images", workerID)
        if err != nil {
            time.Sleep(500 * time.Millisecond)
            continue
        }
        
        task := message.Data.(ImageTask)
        if processImage(task) {
            queue.Acknowledge(message.ID)
        } else {
            queue.Reject(message.ID, "processing failed")
        }
    }
}

4. 📊 Analytics Event Collection

Collect and process analytics events with batching and priority handling.

type AnalyticsEvent struct {
    UserID    string                 `json:"user_id"`
    Event     string                 `json:"event"`
    Properties map[string]interface{} `json:"properties"`
    Timestamp time.Time              `json:"timestamp"`
}

func setupAnalyticsQueue() *lightqueue.Queue {
    config := &lightqueue.QueueConfig{
        Name:           "analytics",
        DataDir:        "./analytics_data",
        BufferSize:     2000, // High volume
        FlushInterval:  10 * time.Second,
        MaxRetries:     2,
        MessageTTL:     6 * time.Hour,
    }
    
    queue, _ := lightqueue.NewQueue(config)
    return queue
}

func trackEvent(queue *lightqueue.Queue, event AnalyticsEvent) {
    priority := 1 // Default priority
    
    // Higher priority for conversion events
    if event.Event == "purchase" || event.Event == "signup" {
        priority = 7
    }
    
    queue.Publish("analytics", event, priority)
}

func processAnalytics(queue *lightqueue.Queue) {
    batch := make([]AnalyticsEvent, 0, 50)
    ticker := time.NewTicker(30 * time.Second)
    
    for {
        select {
        case <-ticker.C:
            if len(batch) > 0 {
                sendToAnalyticsDB(batch)
                batch = batch[:0]
            }
            
        default:
            message, err := queue.Consume("analytics", "analytics-processor")
            if err != nil {
                time.Sleep(100 * time.Millisecond)
                continue
            }
            
            event := message.Data.(AnalyticsEvent)
            batch = append(batch, event)
            queue.Acknowledge(message.ID)
            
            // Send batch when full
            if len(batch) >= 50 {
                sendToAnalyticsDB(batch)
                batch = batch[:0]
            }
        }
    }
}

5. 🤖 AI/ML Model Training Queue

Manage AI model training jobs with priority and resource allocation.

type TrainingJob struct {
    JobID       string            `json:"job_id"`
    ModelType   string            `json:"model_type"`
    DatasetPath string            `json:"dataset_path"`
    Parameters  map[string]interface{} `json:"parameters"`
    UserID      string            `json:"user_id"`
    GPURequired bool              `json:"gpu_required"`
}

func setupTrainingQueue() *lightqueue.Queue {
    config := &lightqueue.QueueConfig{
        Name:           "ml_training",
        DataDir:        "./training_data",
        BufferSize:     100,
        FlushInterval:  30 * time.Second,
        MaxRetries:     2,
        MessageTTL:     48 * time.Hour, // Long-running jobs
    }
    
    queue, _ := lightqueue.NewQueue(config)
    return queue
}

func submitTrainingJob(queue *lightqueue.Queue, job TrainingJob) {
    priority := 3 // Default priority
    
    // Higher priority for premium users or urgent models
    if job.UserID == "premium_user" || job.ModelType == "critical_model" {
        priority = 9
    }
    
    // Lower priority for experimental jobs
    if job.ModelType == "experimental" {
        priority = 1
    }
    
    queue.Publish("training", job, priority)
}

func trainingWorker(queue *lightqueue.Queue) {
    for {
        message, err := queue.Consume("training", "gpu-worker-1")
        if err != nil {
            time.Sleep(5 * time.Second)
            continue
        }
        
        job := message.Data.(TrainingJob)
        
        if trainModel(job) {
            queue.Acknowledge(message.ID)
        } else {
            queue.Reject(message.ID, "training failed")
        }
    }
}

Configuration

QueueConfig Options

type QueueConfig struct {
    Name           string        // Queue name (required)
    DataDir        string        // Directory for SQLite database
    BufferSize     int           // In-memory buffer size before flush
    FlushInterval  time.Duration // How often to flush to database
    MaxRetries     int           // Max retry attempts for failed messages
    RetryDelay     time.Duration // Delay between retries
    MessageTTL     time.Duration // Time-to-live for messages
    EnableMetrics  bool          // Enable statistics collection
}

Default Configuration

func DefaultQueueConfig() *QueueConfig {
    return &QueueConfig{
        Name:           "default_queue",
        DataDir:        "./queue_data",
        BufferSize:     100,
        FlushInterval:  5 * time.Second,
        MaxRetries:     3,
        RetryDelay:     1 * time.Minute,
        MessageTTL:     24 * time.Hour,
        EnableMetrics:  true,
    }
}

API Reference

Core Methods

Publishing Messages

// Publish a message with priority
err := queue.Publish(topic, data, priority)

Consuming Messages

// Consume next highest priority message
message, err := queue.Consume(topic, consumerName)

Message Acknowledgment

// Acknowledge successful processing
err := queue.Acknowledge(messageID)

// Reject message (will retry or go to dead letter)
err := queue.Reject(messageID, reason)

Queue Management

// Get queue statistics
stats := queue.GetStats()

// Force flush to database
err := queue.Flush()

// Cleanup expired messages
err := queue.CleanupExpiredMessages()

// Graceful shutdown
queue.Stop()

Message Structure

type Message struct {
    ID          string      `json:"id"`
    Topic       string      `json:"topic"`
    Data        interface{} `json:"data"`
    Priority    int         `json:"priority"`
    Timestamp   time.Time   `json:"timestamp"`
    Attempts    int         `json:"attempts"`
    MaxAttempts int         `json:"max_attempts"`
    Status      MessageStatus `json:"status"`
    Consumer    string      `json:"consumer"`
    ExpiresAt   *time.Time  `json:"expires_at"`
}

Monitoring and Statistics

stats := queue.GetStats()
fmt.Printf("Pending: %d\n", stats.PendingCount)
fmt.Printf("Processing: %d\n", stats.ProcessingCount)
fmt.Printf("Completed: %d\n", stats.CompletedCount)
fmt.Printf("Failed: %d\n", stats.FailedCount)
fmt.Printf("Retry Count: %d\n", stats.RetryCount)

Best Practices

1. Priority Assignment

• 10: Critical system messages
• 7-9: High priority user actions
• 4-6: Normal business operations
• 1-3: Background tasks and cleanup

2. Error Handling

message, err := queue.Consume("topic", "worker")
if err != nil {
    // Handle consumption errors
    log.Printf("Failed to consume: %v", err)
    time.Sleep(1 * time.Second)
    continue
}

// Process message
if err := processMessage(message); err != nil {
    // Reject on processing failure (will retry)
    queue.Reject(message.ID, err.Error())
} else {
    // Acknowledge on success
    queue.Acknowledge(message.ID)
}

3. Graceful Shutdown

func main() {
    queue, _ := lightqueue.NewQueue(config)
    
    // Handle shutdown signals
    c := make(chan os.Signal, 1)
    signal.Notify(c, os.Interrupt, syscall.SIGTERM)
    
    go func() {
        <-c
        log.Println("Shutting down gracefully...")
        queue.Stop()
        os.Exit(0)
    }()
    
    // Your application logic
}

4. Multiple Workers

// Start multiple consumer workers
for i := 0; i < numWorkers; i++ {
    go func(workerID int) {
        for {
            message, err := queue.Consume("work", fmt.Sprintf("worker-%d", workerID))
            if err != nil {
                time.Sleep(time.Second)
                continue
            }
            
            // Process message
            processWork(message)
            queue.Acknowledge(message.ID)
        }
    }(i)
}

Performance Considerations

• Batch Size: Larger BufferSize improves write performance but uses more memory
• Flush Interval: Shorter intervals provide better durability, longer intervals improve performance
• Workers: Scale consumers based on processing time and throughput requirements
• TTL: Set appropriate message TTL to prevent database bloat

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

License

This project is licensed under the MIT License - see the LICENSE file for details.