Zaiets.EventBus.Lite 1.0.0

Zaiets.EventBus.Lite

In-process event bus for .NET 10 — lightweight pub/sub, async handlers, configurable retry, dead letter queue, and a composable middleware pipeline. Zero external dependencies beyond Microsoft.Extensions.*.

Installation

dotnet add package Zaiets.EventBus.Lite

Quick Start

1. Define an event

using Zaiets.EventBus.Lite.Abstractions;

public sealed record OrderPlaced(Guid OrderId, decimal Total) : EventBase;

2. Write a handler

using Zaiets.EventBus.Lite.Abstractions;

public sealed class SendConfirmationEmailHandler : IEventHandler<OrderPlaced>
{
    private readonly IEmailService _email;

    public SendConfirmationEmailHandler(IEmailService email) => _email = email;

    public async Task HandleAsync(OrderPlaced @event, CancellationToken ct)
    {
        await _email.SendAsync(@event.OrderId, ct);
    }
}

3. Register with DI

// Program.cs
builder.Services
    .AddEventBus(builder.Configuration)
    .AddHandler<OrderPlaced, SendConfirmationEmailHandler>()
    .AddMiddleware<LoggingMiddleware>()
    .AddMiddleware<ActivityMiddleware>();

4. Publish

public class OrderService
{
    private readonly IEventBus _bus;

    public OrderService(IEventBus bus) => _bus = bus;

    public async Task PlaceOrderAsync(Order order, CancellationToken ct)
    {
        // ... business logic ...
        await _bus.PublishAsync(new OrderPlaced(order.Id, order.Total), ct);
    }
}

Configuration

Bind from appsettings.json:

{
  "EventBus": {
    "ContinueOnHandlerError": true,
    "MaxRetryAttempts": 2,
    "RetryDelay": "00:00:00.200",
    "ConcurrentHandlers": false,
    "MaxConcurrency": 0,
    "HandlerTimeout": "00:00:30",
    "EnableLogging": true
  }
}

Or configure inline:

builder.Services
    .AddEventBus(configure: opts =>
    {
        opts.MaxRetryAttempts    = 3;
        opts.ConcurrentHandlers  = true;
        opts.MaxConcurrency      = 4;
        opts.HandlerTimeout      = TimeSpan.FromSeconds(10);
    });

Assembly Scanning

Instead of registering handlers one by one, scan an assembly:

builder.Services
    .AddEventBus()
    .AddHandlersFromAssembly(typeof(OrderPlaced).Assembly);

Inline Subscriptions

For lightweight wiring without a dedicated class:

// Persistent subscription — dispose token to unsubscribe
using var token = bus.Subscribe<OrderPlaced>(async (evt, ct) =>
{
    Console.WriteLine($"Order {evt.OrderId} placed");
    await Task.CompletedTask;
});

// One-time subscription — automatically removed after first fire
bus.SubscribeOnce<OrderPlaced>(async (evt, ct) =>
{
    Console.WriteLine("First order ever!");
    await Task.CompletedTask;
});

Middleware Pipeline

Middlewares wrap every publish call. They execute in registration order (first added = outermost).

Built-in middlewares

Custom middleware

using Zaiets.EventBus.Lite.Abstractions;

public sealed class CorrelationMiddleware : IEventMiddleware
{
    public async Task InvokeAsync(IEvent @event, EventMiddlewareDelegate next, CancellationToken ct)
    {
        using (LogContext.PushProperty("CorrelationId", @event.EventId))
        {
            await next(@event, ct);
        }
    }
}

Register it:

.AddMiddleware<CorrelationMiddleware>()

Dead Letter Queue

Failed events (after all retries) are routed to IDeadLetterQueue.

Reading the DLQ

public class DeadLetterMonitor
{
    private readonly IDeadLetterQueue _dlq;

    public DeadLetterMonitor(IDeadLetterQueue dlq) => _dlq = dlq;

    public async Task PrintAsync()
    {
        var letters = await _dlq.GetAllAsync();
        foreach (var dl in letters)
            Console.WriteLine($"[{dl.FailedAt}] {dl.Event.GetType().Name} — {dl.Exception.Message}");
    }
}

Custom DLQ (e.g., write to a database)

public class SqlDeadLetterQueue : IDeadLetterQueue
{
    // ... your EF Core / Dapper implementation ...
}

// Registration — call UseDeadLetterQueue BEFORE AddHandler calls
builder.Services
    .AddEventBus()
    .UseDeadLetterQueue<SqlDeadLetterQueue>();

OpenTelemetry Integration

Wire up ActivityMiddleware and listen on the named source:

using OpenTelemetry.Trace;

builder.Services.AddOpenTelemetry().WithTracing(tracing =>
{
    tracing
        .AddSource(Zaiets.EventBus.Lite.Middleware.ActivityMiddleware.Source.Name)
        .AddJaegerExporter();
});

Publish Multiple Events

var events = orders.Select(o => new OrderPlaced(o.Id, o.Total));
await bus.PublishManyAsync(events, cancellationToken);

DataAnnotations Validation

Add ValidationMiddleware and decorate your events:

using System.ComponentModel.DataAnnotations;

public sealed record OrderPlaced(
    [Required] Guid OrderId,
    [Range(0.01, double.MaxValue)] decimal Total) : EventBase;

Publishing an invalid event throws EventValidationException with a list of ValidationResult errors before any handler runs.

API Reference

IEventBus

IEventHandler<TEvent>

Implement and register to handle TEvent via DI.

IEventMiddleware

Implement InvokeAsync(event, next, ct) to create cross-cutting pipeline behaviour.

IDeadLetterQueue

License

MIT © 2025 Vladyslav Zaiets