ControlTowR 1.0.0-beta.2

ControlTowR

A MediatR-compatible CQRS/Mediator library for .NET — built from scratch, targeting netstandard2.1.

Installation

dotnet add package ControlTowR.Extensions.Microsoft.DependencyInjection

This pulls in the core ControlTowR package automatically. If you only need the interfaces (e.g. for a custom DI integration), install the core package directly:

dotnet add package ControlTowR

Getting Started

1. Register with DI

services.AddControlTowR(typeof(MyHandler).Assembly);

2. Define a request and handler

// Query (returns a value)
public record GetUserQuery(int Id) : IRequest<User>;

public class GetUserQueryHandler : IRequestHandler<GetUserQuery, User>
{
    public Task<User> Handle(GetUserQuery request, CancellationToken cancellationToken)
        => userRepository.GetByIdAsync(request.Id, cancellationToken);
}

// Command (no return value)
public record DeleteUserCommand(int Id) : IRequest;

public class DeleteUserCommandHandler : IRequestHandler<DeleteUserCommand>
{
    public async Task Handle(DeleteUserCommand request, CancellationToken cancellationToken)
        => await userRepository.DeleteAsync(request.Id, cancellationToken);
}

3. Send requests

public class UserController(IMediator mediator)
{
    public Task<User> GetUser(int id, CancellationToken ct)
        => mediator.Send(new GetUserQuery(id), ct);

    public Task DeleteUser(int id, CancellationToken ct)
        => mediator.Send(new DeleteUserCommand(id), ct);
}

Notifications

Broadcast a message to multiple handlers:

public record UserCreated(string Username) : INotification;

public class SendWelcomeEmail : INotificationHandler<UserCreated>
{
    public Task Handle(UserCreated notification, CancellationToken cancellationToken)
        => emailService.SendWelcomeAsync(notification.Username, cancellationToken);
}

// Publish
await mediator.Publish(new UserCreated("alice"), cancellationToken);

Handlers are invoked sequentially by default. Multiple handlers for the same notification are all invoked.

Pipeline Behaviors

Add cross-cutting concerns (logging, validation, caching) without modifying handlers:

public class LoggingBehavior<TRequest, TResponse>(ILogger<LoggingBehavior<TRequest, TResponse>> logger)
    : IPipelineBehavior<TRequest, TResponse> where TRequest : notnull
{
    public async Task<TResponse> Handle(TRequest request, RequestHandlerDelegate<TResponse> next, CancellationToken ct)
    {
        logger.LogInformation("Handling {Request}", typeof(TRequest).Name);
        var response = await next(ct);
        logger.LogInformation("Handled {Request}", typeof(TRequest).Name);
        return response;
    }
}

Register as an open generic to apply to all requests:

services.AddTransient(typeof(IPipelineBehavior<,>), typeof(LoggingBehavior<,>));

Or let assembly scanning handle it automatically if the behavior is in a scanned assembly.

Behaviors execute in registration order — first registered is the outermost wrapper.

Streaming

For handlers that produce multiple results over time:

public record SearchProductsStream(string Query) : IStreamRequest<Product>;

public class SearchProductsStreamHandler : IStreamRequestHandler<SearchProductsStream, Product>
{
    public async IAsyncEnumerable<Product> Handle(
        SearchProductsStream request,
        [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        await foreach (var product in repository.StreamAsync(request.Query, cancellationToken))
            yield return product;
    }
}

// Consume
await foreach (var product in mediator.CreateStream(new SearchProductsStream("laptop"), ct))
{
    Console.WriteLine(product.Name);
}

Always decorate cancellationToken with [EnumeratorCancellation] in stream handler implementations.

Exception Handling

Exception Handler (suppress and return fallback)

public class OutOfStockHandler
    : IRequestExceptionHandler<PlaceOrderCommand, OrderResult, InsufficientStockException>
{
    public Task Handle(PlaceOrderCommand request, InsufficientStockException exception,
        RequestExceptionHandlerState<OrderResult> state, CancellationToken cancellationToken)
    {
        state.SetHandled(new OrderResult(Success: false, Reason: "Item out of stock"));
        return Task.CompletedTask;
    }
}

Exception Action (side-effect, exception still propagates)

public class DatabaseExceptionLogger<TRequest>
    : IRequestExceptionAction<TRequest, DatabaseException> where TRequest : notnull
{
    public Task Execute(TRequest request, DatabaseException exception, CancellationToken cancellationToken)
    {
        logger.LogError(exception, "Database error in {Request}", typeof(TRequest).Name);
        return Task.CompletedTask;
    }
}

Register the built-in behavior to enable exception handling:

services.AddTransient(typeof(IPipelineBehavior<,>), typeof(RequestExceptionProcessorBehavior<,>));

Assembly Scanning

AddControlTowR scans assemblies and auto-registers all handlers, behaviors, and exception types:

// Single assembly
services.AddControlTowR(typeof(MyHandler).Assembly);

// Multiple assemblies
services.AddControlTowR([typeof(MyHandler).Assembly, typeof(OtherHandler).Assembly]);

// With type filter (useful to exclude test fixtures)
services.AddControlTowR(config => config
    .AddAssembly(typeof(MyHandler).Assembly)
    .Where(t => t.Namespace?.StartsWith("MyApp") == true));

Registered lifetimes: | Type | Lifetime | |------|----------| | IRequestHandler, INotificationHandler, IStreamRequestHandler | Scoped | | IPipelineBehavior, IStreamPipelineBehavior, IRequestExceptionHandler, IRequestExceptionAction | Transient |

Feedback & Bug Reports

Found a bug or have a feature request? Please open an issue on GitHub.

When reporting a bug, include:

• The version of ControlTowR you're using
• A minimal reproduction (request type, handler, registration code)
• The expected vs. actual behaviour

License

GPLv3