Philiprehberger.EventBus 0.6.0

Philiprehberger.EventBus

In-process publish/subscribe event bus with middleware pipeline, dead-letter queue, event replay, and Microsoft DI integration.

Installation

dotnet add package Philiprehberger.EventBus

Usage

using Philiprehberger.EventBus;

var bus = new EventBus();

using var subscription = bus.Subscribe<OrderPlaced>(async (e, ct) =>
{
    Console.WriteLine($"Order {e.OrderId} placed");
});

await bus.PublishAsync(new OrderPlaced(OrderId: 42));

record OrderPlaced(int OrderId);

Publish and Subscribe

using Philiprehberger.EventBus;

var bus = new EventBus();

// Subscribe returns an IDisposable — dispose it to unsubscribe
using var sub = bus.Subscribe<UserRegistered>(async (e, ct) =>
{
    await SendWelcomeEmailAsync(e.Email, ct);
});

// Publish fires all handlers concurrently
await bus.PublishAsync(new UserRegistered("user@example.com"));

record UserRegistered(string Email);

Handler Priority

using Philiprehberger.EventBus;

var bus = new EventBus();

// Lower priority number executes first
bus.Subscribe<OrderPlaced>((e, ct) =>
{
    Console.WriteLine("Validate order");
    return Task.CompletedTask;
}, priority: 10);

bus.Subscribe<OrderPlaced>((e, ct) =>
{
    Console.WriteLine("Send confirmation email");
    return Task.CompletedTask;
}, priority: 20);

await bus.PublishAsync(new OrderPlaced(1));
// Output: Validate order, then Send confirmation email

record OrderPlaced(int OrderId);

Handler Filtering

using Philiprehberger.EventBus;

var bus = new EventBus();

// Only handle high-value orders
bus.Subscribe<OrderPlaced>(
    (e, ct) =>
    {
        Console.WriteLine($"High-value order: {e.OrderId}");
        return Task.CompletedTask;
    },
    filter: e => e.Total > 1000);

await bus.PublishAsync(new OrderPlaced(1, 500));   // Skipped
await bus.PublishAsync(new OrderPlaced(2, 2000));  // Handled

record OrderPlaced(int OrderId, decimal Total);

Error Handling

using Philiprehberger.EventBus;

var bus = new EventBus(new EventBusOptions
{
    ThrowOnHandlerError = false,
    OnHandlerError = ex => Console.Error.WriteLine($"Handler failed: {ex.Message}")
});

bus.Subscribe<OrderPlaced>((_, _) => throw new InvalidOperationException("oops"));

await bus.PublishAsync(new OrderPlaced(1));
// Logs "Handler failed: oops" without propagating the exception

record OrderPlaced(int OrderId);

Handler Timeout

using Philiprehberger.EventBus;

var bus = new EventBus(new EventBusOptions
{
    ThrowOnHandlerError = true,
    HandlerTimeout = TimeSpan.FromSeconds(5)
});

bus.Subscribe<OrderPlaced>(async (e, ct) =>
{
    await ProcessOrderAsync(e.OrderId, ct);
});

// Throws TimeoutException if the handler exceeds 5 seconds
await bus.PublishAsync(new OrderPlaced(1));

record OrderPlaced(int OrderId);

Dead-Letter Queue

using Philiprehberger.EventBus;

var bus = new EventBus(new EventBusOptions
{
    ThrowOnHandlerError = false,
    OnDeadLetter = (evt, ex) =>
        Console.Error.WriteLine($"Dead letter: {evt.GetType().Name} failed with {ex.Message}")
});

bus.Subscribe<OrderPlaced>((_, _) => throw new InvalidOperationException("payment failed"));

await bus.PublishAsync(new OrderPlaced(1));
// Logs "Dead letter: OrderPlaced failed with payment failed"

record OrderPlaced(int OrderId);

Event Replay

using Philiprehberger.EventBus;

var bus = new EventBus();
bus.EnableHistory(maxEvents: 100);

bus.Subscribe<OrderPlaced>((e, _) =>
{
    Console.WriteLine($"Order {e.OrderId}");
    return Task.CompletedTask;
});

await bus.PublishAsync(new OrderPlaced(1));
await bus.PublishAsync(new OrderPlaced(2));
await bus.PublishAsync(new OrderPlaced(3));

// Re-publishes the 2 most recent events (OrderPlaced 2, then 3)
await bus.ReplayLastAsync(2);

record OrderPlaced(int OrderId);

Middleware

using Philiprehberger.EventBus;

var bus = new EventBus();

// Add logging middleware
bus.Use(async (context, next) =>
{
    Console.WriteLine($"Before: {context.EventType.Name}");
    await next();
    Console.WriteLine($"After: {context.EventType.Name}");
});

// Add timing middleware
bus.Use(async (context, next) =>
{
    var sw = System.Diagnostics.Stopwatch.StartNew();
    await next();
    sw.Stop();
    Console.WriteLine($"Handler took {sw.ElapsedMilliseconds}ms");
});

bus.Subscribe<OrderPlaced>((e, _) =>
{
    Console.WriteLine($"Processing order {e.OrderId}");
    return Task.CompletedTask;
});

await bus.PublishAsync(new OrderPlaced(1));

record OrderPlaced(int OrderId);

One-Time Subscription

using Philiprehberger.EventBus;

var bus = new EventBus();

// Handler is automatically unsubscribed after the first matching event
bus.SubscribeOnce<OrderPlaced>((e, _) =>
{
    Console.WriteLine($"First order placed: {e.OrderId}");
    return Task.CompletedTask;
});

await bus.PublishAsync(new OrderPlaced(1)); // Handled
await bus.PublishAsync(new OrderPlaced(2)); // Ignored — already unsubscribed

record OrderPlaced(int OrderId);

Await Next Event

using Philiprehberger.EventBus;

var bus = new EventBus();

// Start waiting before the event is published
var waitTask = bus.WaitForAsync<OrderPlaced>(
    filter: e => e.Total > 1000);

// Publish events from another part of the application
await bus.PublishAsync(new OrderPlaced(1, 500));   // Skipped by filter
await bus.PublishAsync(new OrderPlaced(2, 2000));  // Completes the wait

var order = await waitTask;
Console.WriteLine($"High-value order: {order.OrderId}");

record OrderPlaced(int OrderId, decimal Total);

Check Subscribers

using Philiprehberger.EventBus;

var bus = new EventBus();

Console.WriteLine(bus.HasSubscribers<OrderPlaced>()); // False
Console.WriteLine(bus.GetSubscriberCount<OrderPlaced>()); // 0

using var sub = bus.Subscribe<OrderPlaced>((_, _) => Task.CompletedTask);
Console.WriteLine(bus.HasSubscribers<OrderPlaced>()); // True
Console.WriteLine(bus.GetSubscriberCount<OrderPlaced>()); // 1

record OrderPlaced(int OrderId);

Bulk Unsubscribe

using Philiprehberger.EventBus;

var bus = new EventBus();

bus.Subscribe<OrderPlaced>((e, _) => { Console.WriteLine(e.OrderId); return Task.CompletedTask; });
bus.Subscribe<OrderPlaced>((e, _) => { Console.WriteLine("Audit: " + e.OrderId); return Task.CompletedTask; });

// Remove all handlers for a specific event type
bus.UnsubscribeAll<OrderPlaced>();

// Or remove all handlers for all event types
bus.UnsubscribeAll();

record OrderPlaced(int OrderId);

Inspect Event History

using Philiprehberger.EventBus;

var bus = new EventBus();
bus.EnableHistory(100);

await bus.PublishAsync(new OrderPlaced(1));
await bus.PublishAsync(new OrderPlaced(2));
await bus.PublishAsync(new OrderPlaced(3));

// Read recorded events without re-triggering handlers
IReadOnlyList<object> history = bus.GetHistory();
foreach (var evt in history)
{
    Console.WriteLine(((OrderPlaced)evt).OrderId);
}
// Output: 1, 2, 3

// Typed history filtered to a specific event type
IReadOnlyList<OrderPlaced> orders = bus.GetHistory<OrderPlaced>();

record OrderPlaced(int OrderId);

Toggle History

using Philiprehberger.EventBus;

var bus = new EventBus();

if (!bus.IsHistoryEnabled)
{
    bus.EnableHistory(100);
}

// ...later, when telemetry is no longer needed...
bus.DisableHistory();

DI Registration

using Philiprehberger.EventBus;

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddEventBus(options =>
{
    options.ThrowOnHandlerError = true;
    options.MaxConcurrency = 4;
    options.HandlerTimeout = TimeSpan.FromSeconds(10);
    options.OnHandlerError = ex => Console.Error.WriteLine(ex);
});

var app = builder.Build();

Handler Classes

using Philiprehberger.EventBus;

public record OrderShipped(int OrderId, string TrackingNumber);

public class OrderShippedHandler : IEventHandler<OrderShipped>
{
    public async Task HandleAsync(OrderShipped @event, CancellationToken ct)
    {
        await NotifyCustomerAsync(@event.OrderId, @event.TrackingNumber, ct);
    }
}

// Subscribe an IEventHandler<T> instance directly (no DI required)
var bus = new EventBus();
var handler = new OrderShippedHandler();
using var sub = bus.Subscribe<OrderShipped>(handler);

API

IEventBus

IEventHandler<T>

EventBusOptions

Subscribe<T> Parameters

EventContext

ServiceCollectionExtensions

Development

dotnet build src/Philiprehberger.EventBus.csproj --configuration Release

Support

If you find this project useful:

⭐ Star the repo

🐛 Report issues

💡 Suggest features

❤️ Sponsor development

🌐 All Open Source Projects

💻 GitHub Profile

🔗 LinkedIn Profile

License

MIT