Valir

A modular .NET distributed job queue framework with Redis backend, supporting at-least-once delivery, priority queues, transactional outbox, and event-driven architectures via Kafka/RabbitMQ/Azure Service Bus.

Features • Quick Start • Packages • Documentation • Contributing

---

Features

Feature	Description
✅ At-Least-Once Delivery	Idempotency keys for reliable processing
✅ Priority Queues	Higher priority jobs processed first
✅ Batch Operations	Enqueue thousands of jobs efficiently
✅ Graceful Shutdown	Drain mode for zero job loss
✅ Transactional Outbox	Atomic job creation with your DB
✅ Distributed Locks	Redis-backed coordination
✅ Rate Limiting	Sliding window algorithm
✅ OpenTelemetry	Native tracing support
✅ TUI Dashboard	Interactive worker console

Quick Start

Installation

# Core packages
dotnet add package Valir.Redis
dotnet add package Valir.AspNet

# Event Bus (choose one)
dotnet add package Valir.Brokers.Kafka
dotnet add package Valir.Brokers.RabbitMQ
dotnet add package Valir.Brokers.AzureSB

Producer (Web API)

// Program.cs
builder.Services.AddValir(options =>
{
    options.RedisConnectionString = "localhost:6379";
});

// Enqueue a job
app.MapPost("/send-email", async (IJobQueue queue, EmailRequest req) =>
{
    byte[] payload = JsonSerializer.SerializeToUtf8Bytes(req);
    string jobId = await queue.EnqueueAsync(
        type: "send-email",
        payload: payload,
        priority: 5,
        idempotencyKey: $"email-{req.UserId}"
    );
    return Results.Ok(new { JobId = jobId });
});

Consumer (Worker)

// Define your job handler
public class EmailJobHandler : IJobHandler<EmailRequest>
{
    public async Task HandleAsync(EmailRequest request, JobContext context)
    {
        // Process the job
        await SendEmailAsync(request, context.CancellationToken);
        
        Console.WriteLine($"Email sent to {request.Email}");
    }
}

// Program.cs - Register and run
builder.Services.AddValir(options =>
{
    options.RedisConnectionString = "localhost:6379";
    options.Concurrency = 4;
});

builder.Services.AddSingleton<IJobHandler<EmailRequest>, EmailJobHandler>();

// Start worker runtime
var worker = app.Services.GetRequiredService<WorkerRuntime>();
await worker.StartAsync(CancellationToken.None);

Or use the sample worker with TUI:

dotnet run --project samples/Valir.Sample.Worker -- --redis localhost:6379 --concurrency 4

Packages

Package	Description	NuGet
Valir.Abstractions	Core interfaces
Valir.Core	Worker runtime, retry policies
Valir.Redis	Redis job queue, distributed lock
Valir.AspNet	ASP.NET Core integration
Valir.EntityFrameworkCore	Transactional Outbox pattern
Valir.Brokers.Kafka	Apache Kafka adapter
Valir.Brokers.RabbitMQ	RabbitMQ adapter
Valir.Brokers.AzureSB	Azure Service Bus adapter

Architecture

graph LR
    subgraph Producer
        A[Web API]
    end
    
    subgraph Storage
        B[(Redis<br/>Job Queue)]
    end
    
    subgraph Consumer
        C[Worker]
        D[Handler<br/>Your Code]
    end
    
    subgraph Events
        E[Event Bus<br/>Kafka/RMQ/Azure]
    end
    
    A -->|Enqueue| B
    B -->|Claim| C
    C --> D
    A -->|Publish| E

Loading

Configuration

services.AddValir(options =>
{
    // Redis connection
    options.RedisConnectionString = "localhost:6379";
    options.KeyPrefix = "valir:";
    
    // Worker settings
    options.Concurrency = 4;
    options.DefaultMaxAttempts = 3;
    options.RetryBaseDelay = TimeSpan.FromSeconds(10);
    
    // Timeouts
    options.DefaultVisibilityTimeout = TimeSpan.FromSeconds(30);
    options.ShutdownTimeout = TimeSpan.FromSeconds(30);
});

Event Bus Adapters

Kafka

services.AddValirKafka(options =>
{
    options.BootstrapServers = "localhost:9092";
    options.GroupId = "my-service";
    options.EnableAutoCommit = false; // At-least-once
});

RabbitMQ

services.AddValirRabbitMQ(options =>
{
    options.HostName = "localhost";
    options.UserName = "guest";
    options.Password = "guest";
    options.ExchangeName = "valir.events";
});

Azure Service Bus

services.AddValirAzureServiceBus(options =>
{
    options.ConnectionString = "Endpoint=sb://...";
    options.MaxConcurrentCalls = 10;
});

Transactional Outbox

Ensure job creation is atomic with your database transaction:

// Register outbox
services.AddValirOutbox<AppDbContext>();

// In your service
public async Task CreateOrderAsync(Order order)
{
    _context.Orders.Add(order);
    
    // Job is written to outbox table (same transaction)
    await _outboxQueue.EnqueueAsync("process-order", payload);
    
    await _context.SaveChangesAsync(); // Atomic!
}
// Background processor pushes to Redis automatically

Documentation

• Getting Started
• Configuration
• Event Bus
• Transactional Outbox
• API Reference
• Docker

Contributing

Contributions are welcome! Please read our Contributing Guide for details.

1. Fork the repository
2. Create your feature branch (git checkout -b feature/amazing-feature)
3. Commit your changes (git commit -m 'Add amazing feature')
4. Push to the branch (git push origin feature/amazing-feature)
5. Open a Pull Request

License

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

---

Made with ❤️ by Taiizor