QueueEngine 1.0.0

QueueEngine

A lightweight, production-ready queue engine for .NET 10 with support for SQLite and PostgreSQL.

📖 For detailed usage guide, see DEVELOPER_GUIDE.md

Features

• Multiple Queues: Support for multiple named queues with individual concurrency and rate limiting
• Dual Database Support: SQLite for development, PostgreSQL for production
• Job Handlers: Strongly-typed job handlers with automatic JSON serialization
• CRON Scheduling: Schedule recurring jobs using CRON expressions
• Rate Limiting: Per-queue rate limiting to control throughput
• Retry Mechanism: Automatic retry with exponential backoff for failed jobs
• Dead Letter Queue: Move permanently failed jobs for analysis
• Job Cancellation: Cancel pending or running jobs
• Bulk Enqueue: Enqueue multiple jobs at once
• Job Priority: Prioritize jobs within a queue (higher priority jobs processed first)
• Job Progress: Handlers can report progress (0-100%) during execution
• Pause/Resume Queue: Pause or resume individual queues without stopping the engine
• Queue Statistics: Get detailed statistics per queue (pending, running, done, failed, cancelled)
• Health Checks: ASP.NET Core health check integration
• Dashboard: Web-based dashboard for monitoring queues and workers
• Dependency Injection: Full DI support for integration with ASP.NET Core
• Logging: Structured logging using Microsoft.Extensions.Logging
• Testable: Interfaces for all core components for easy unit testing
• Distributed Cluster: Worker registration, heartbeats, and job stealing support

Installation

dotnet add package QueueEngine

Or install dependencies manually:

dotnet add package Dapper
dotnet add package Microsoft.Data.Sqlite
dotnet add package Npgsql
dotnet add package Microsoft.Extensions.DependencyInjection
dotnet add package Microsoft.Extensions.Hosting.Abstractions
dotnet add package Microsoft.Extensions.Logging
dotnet add package Microsoft.Extensions.Logging.Console
dotnet add package Microsoft.Extensions.Diagnostics.HealthChecks

Dashboard

Run the built-in dashboard to monitor your queues in real-time:

cd src/QueueEngine.Dashboard
dotnet run

Then open http://localhost:5000 in your browser.

The dashboard provides:

• Real-time job statistics (pending, running, done, failed)
• Queue management (pause/resume)
• Worker status monitoring

Benchmarks

Run performance benchmarks to test throughput:

cd benchmarks/QueueEngine.Benchmarks
dotnet run

Quick Start

Console Application

using Microsoft.Extensions.Logging;
using QueueEngine.Config;
using QueueEngine.Core;
using QueueEngine.Data;
using QueueEngine.Workers;

var loggerFactory = LoggerFactory.Create(builder =>
{
    builder.AddConsole();
    builder.SetMinimumLevel(LogLevel.Information);
});

var logger = loggerFactory.CreateLogger<QueueEngine.Core.QueueEngine>();

var options = new QueueEngineOptions
{
    ConnectionString = "Data Source=queue.db",
    DatabaseProvider = "sqlite",
    Queues = new Dictionary<string, QueueOptions>
    {
        ["default"] = new QueueOptions { Concurrency = 2, RateLimitPerSecond = 10 }
    }
};

options.Validate();

var repository = new JobRepository(options);
var workerPool = new QueueWorkerPool(repository, new Dictionary<string, IJobHandler>(), options, loggerFactory.CreateLogger<QueueWorkerPool>());
var engine = new QueueEngine.Core.QueueEngine(repository, workerPool, options, logger);

engine.RegisterHandler(new MyJobHandler());

await engine.EnqueueAsync("my-job", new MyPayload("Hello World"));
await engine.StartAsync();

await Task.Delay(5000);

await engine.StopAsync();

ASP.NET Core Integration

// Program.cs
builder.Services
    .AddQueueEngine(opts =>
    {
        opts.ConnectionString = builder.Configuration.GetConnectionString("Queue")!;
        opts.DatabaseProvider = "postgres";
        opts.Queues["default"] = new QueueOptions
        {
            Concurrency = 4,
            RateLimitPerSecond = 20
        };
    })
    .AddJobHandler<EmailJobHandler>()
    .AddQueueEngineHostedService();

// Add health checks
builder.Services.AddHealthChecks()
    .AddQueueEngineHealthCheck();

Creating Job Handlers

Step 1: Define Your Payload

public record SendEmailPayload(string To, string Subject, string Body);

Step 2: Create the Handler

using QueueEngine.Models;
using QueueEngine.Workers;

public class EmailJobHandler : JobHandler<SendEmailPayload>
{
    public override string JobType => "send-email";

    protected override async Task HandleAsync(QueueJob job, SendEmailPayload payload, CancellationToken ct)
    {
        Console.WriteLine($"Sending email to {payload.To}");
        await Task.Delay(500, ct);
    }
}

Step 3: Register and Use

engine.RegisterHandler(new EmailJobHandler());
await engine.EnqueueAsync("send-email", new SendEmailPayload("user@example.com", "Hello", "World!"));

API Reference

QueueEngine Methods

// Enqueue a single job
await engine.EnqueueAsync("job-type", payload, "queue-name", scheduledAt: DateTime.UtcNow.AddHours(1));

// Bulk enqueue
await engine.BulkEnqueueAsync(new[] {
    ("send-email", new EmailPayload(...), "default"),
    ("notify-user", new NotificationPayload(...), "critical")
});

// Cancel a job
await engine.CancelJobAsync(jobId);

// Get job status
var job = await engine.GetJobAsync(jobId);

// Move to dead letter
await engine.MoveToDeadLetterAsync(jobId);

// Get dead letter jobs
var deadJobs = await engine.GetDeadLetterJobsAsync("default");

// Get queue stats
var stats = await engine.GetStatsAsync();

Configuration

QueueEngineOptions

QueueOptions

Architecture

QueueEngine/
├── Config/
│   └── QueueEngineOptions.cs       # Configuration classes
├── Core/
│   ├── IQueueEngine.cs             # Engine interface
│   └── QueueEngine.cs              # Main engine implementation
├── Data/
│   ├── IJobRepository.cs           # Repository interface
│   └── JobRepository.cs            # Database operations
├── Models/
│   └── QueueJob.cs                  # Job entity
├── Workers/
│   ├── IJobHandler.cs              # Handler interface
│   ├── IQueueWorkerPool.cs         # Worker pool interface
│   ├── QueueWorkerPool.cs          # Worker management
│   └── RateLimiter.cs              # Rate limiting
├── Handlers/
│   └── ExampleHandlers.cs          # Example handlers
├── QueueEngineExtensions.cs        # DI extensions
├── QueueEngineHealthCheck.cs       # Health check
└── Program.cs                      # Demo application

Security

Input Validation

• Job types and queue names validated (alphanumeric, dash, underscore only)
• Maximum length limits (255 chars job type, 100 chars queue)
• Payload size limited to 1MB

SQL Injection Protection

• Parameterized queries via Dapper
• No string concatenation in SQL

Error Message Sanitization

• Truncated to 500 chars
• Stack traces never exposed

Database Recommendations

Warning: SQLite is not suitable for high-concurrency production. Always use PostgreSQL.

License

MIT