Moderator 0.7.2

Moderator

A high-performance, lightweight in-process mediator implementation for .NET applications. Moderator provides a clean abstraction for request/response patterns, event notifications, streaming operations, and cross-cutting concerns through pipeline behaviors.

📋 Table of Contents

• Features
• Installation
• Getting Started
• Core Concepts
  • Request/Response
  • Notifications
  • Streaming
  • Pipeline Behaviors
• Advanced Usage
• Source Generator
• Performance
• Contributing
• License

✨ Features

• Simple and Explicit Contracts - Clear interfaces for request handlers and notification handlers
• Streaming Support - Built-in support for async streaming operations
• Pipeline Behaviors - Compose cross-cutting concerns with both unary and streaming pipelines
• Zero Value Results - Included Unit type for commands without return values
• Source Generator - Optional compile-time handler discovery with zero runtime reflection
• Minimal Dependencies - Lightweight implementation with no external dependencies
• Thread-Safe - Fully thread-safe mediator implementation
• Testable - Easy to mock and test with clear interface contracts

📦 Installation

Package Manager Console

Install-Package Moderator
Install-Package Moderator.Contracts
Install-Package Moderator.Generator  # Optional but recommended

.NET CLI

dotnet add package Moderator
dotnet add package Moderator.Contracts
dotnet add package Moderator.Generator  # Optional but recommended

Package Reference

<PackageReference Include="Moderator" Version="*" />
<PackageReference Include="Moderator.Contracts" Version="*" />
<PackageReference Include="Moderator.Generator" Version="*" />  

🚀 Getting Started

Basic Setup

Option 1: With Source Generator (Recommended)

using Microsoft.Extensions.DependencyInjection;
using Moderator;

var builder = WebApplication.CreateBuilder(args);

// Single line registration with source generator
builder.Services.AddModerator(config =>
{
    config.DiscoveredTypes = DiscoveredTypes.All;
});

var app = builder.Build();

// Use without specifying generic types
app.MapPost("/users", async (IModerator moderator, CreateUserRequest request) =>
{
    // Type inference from generated extension methods
    var response = await moderator.SendAsync(request);
    await moderator.PublishAsync(new UserCreatedEvent(response.UserId));
    return Results.Ok(response);
});

app.Run();

Option 2: Manual Registration

using Microsoft.Extensions.DependencyInjection;
using Moderator;

var services = new ServiceCollection();

// Register the mediator
services.AddScoped<IModerator, Moderator>();

// Manually register your handlers
services.AddScoped<IRequestHandler<GetUserQuery, UserDto>, GetUserHandler>();
services.AddScoped<INotificationHandler<UserCreatedEvent>, UserCreatedEventHandler>();

// Register pipeline behaviors (order matters!)
services.AddScoped(typeof(IPipelineBehavior<,>), typeof(LoggingBehavior<,>));
services.AddScoped(typeof(IPipelineBehavior<,>), typeof(ValidationBehavior<,>));

var serviceProvider = services.BuildServiceProvider();
var moderator = serviceProvider.GetRequiredService<IModerator>();

Simple Request/Response Example

// Query definition
public sealed record GetUserByIdQuery(Guid UserId);

// Response DTO
public sealed record UserDto(Guid Id, string Name, string Email);

// Handler implementation
public sealed class GetUserByIdHandler : IRequestHandler<GetUserByIdQuery, UserDto>
{
    private readonly IUserRepository _repository;

    public GetUserByIdHandler(IUserRepository repository)
    {
        _repository = repository;
    }

    public async Task<UserDto> HandleAsync(
        GetUserByIdQuery request, 
        CancellationToken cancellationToken = default)
    {
        var user = await _repository.GetByIdAsync(request.UserId, cancellationToken);
        return new UserDto(user.Id, user.Name, user.Email);
    }
}

// Usage with source generator (type inference)
var user = await moderator.SendAsync(
    new GetUserByIdQuery(userId), 
    cancellationToken);

// Usage without source generator (explicit types)
var user = await moderator.SendAsync<GetUserByIdQuery, UserDto>(
    new GetUserByIdQuery(userId), 
    cancellationToken);

📚 Core Concepts

Request/Response

Implement IRequestHandler<TRequest, TResponse> for synchronous request/response patterns:

public interface IRequestHandler<in TRequest, TResponse>
{
    Task<TResponse> HandleAsync(TRequest request, CancellationToken cancellationToken = default);
}

Commands (No Return Value)

For commands that don't return a value, use the Unit type:

public sealed record DeleteUserCommand(Guid UserId);

public sealed class DeleteUserHandler : IRequestHandler<DeleteUserCommand, Unit>
{
    private readonly IUserRepository _repository;

    public DeleteUserHandler(IUserRepository repository)
    {
        _repository = repository;
    }

    public async Task<Unit> HandleAsync(
        DeleteUserCommand request, 
        CancellationToken cancellationToken = default)
    {
        await _repository.DeleteAsync(request.UserId, cancellationToken);
        return Unit.Value;
    }
}

Notifications

Implement INotificationHandler<TNotification> for pub/sub patterns:

// Notification must implement INotification marker interface
public sealed record UserRegisteredEvent(
    Guid UserId, 
    string Email, 
    DateTime RegisteredAt) : INotification;

// Multiple handlers can handle the same notification
public sealed class SendWelcomeEmailHandler : INotificationHandler<UserRegisteredEvent>
{
    private readonly IEmailService _emailService;

    public SendWelcomeEmailHandler(IEmailService emailService)
    {
        _emailService = emailService;
    }

    public async Task HandleAsync(
        UserRegisteredEvent notification, 
        CancellationToken cancellationToken = default)
    {
        await _emailService.SendWelcomeEmailAsync(
            notification.Email, 
            cancellationToken);
    }
}

public sealed class UpdateAnalyticsHandler : INotificationHandler<UserRegisteredEvent>
{
    public async Task HandleAsync(
        UserRegisteredEvent notification, 
        CancellationToken cancellationToken = default)
    {
        // Update analytics
        await Task.CompletedTask;
    }
}

// Publishing notifications
await moderator.PublishAsync(
    new UserRegisteredEvent(userId, email, DateTime.UtcNow), 
    cancellationToken);

Streaming

For progressive data streaming, implement IStreamRequestHandler<TRequest, TResponse>:

public sealed record SearchProductsQuery(
    string SearchTerm, 
    int MaxResults = 100);

public sealed class SearchProductsHandler 
    : IStreamRequestHandler<SearchProductsQuery, ProductSearchResult>
{
    private readonly IProductSearchService _searchService;

    public SearchProductsHandler(IProductSearchService searchService)
    {
        _searchService = searchService;
    }

    public async IAsyncEnumerable<ProductSearchResult> HandleAsync(
        SearchProductsQuery request,
        [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        await foreach (var batch in _searchService.SearchAsync(
            request.SearchTerm, 
            request.MaxResults, 
            cancellationToken))
        {
            foreach (var product in batch)
            {
                yield return new ProductSearchResult(
                    product.Id, 
                    product.Name, 
                    product.Price);
            }
        }
    }
}

// Consuming streams
await foreach (var result in moderator.StreamAsync<SearchProductsQuery, ProductSearchResult>(
    new SearchProductsQuery("laptop", 50), 
    cancellationToken))
{
    Console.WriteLine($"Found: {result.Name} - ${result.Price}");
}

Pipeline Behaviors

Pipeline behaviors wrap handler execution for cross-cutting concerns:

public sealed class TimingBehavior<TRequest, TResponse> 
    : IPipelineBehavior<TRequest, TResponse>
{
    private readonly ILogger<TimingBehavior<TRequest, TResponse>> _logger;

    public TimingBehavior(ILogger<TimingBehavior<TRequest, TResponse>> logger)
    {
        _logger = logger;
    }

    public async Task<TResponse> HandleAsync(
        TRequest request,
        RequestHandlerDelegate<TResponse> next,
        CancellationToken cancellationToken = default)
    {
        var stopwatch = Stopwatch.StartNew();
        
        try
        {
            var response = await next(cancellationToken);
            stopwatch.Stop();
            
            if (stopwatch.ElapsedMilliseconds > 500)
            {
                _logger.LogWarning(
                    "Long running request: {RequestName} ({ElapsedMilliseconds}ms)",
                    typeof(TRequest).Name,
                    stopwatch.ElapsedMilliseconds);
            }
            
            return response;
        }
        catch (Exception ex)
        {
            stopwatch.Stop();
            _logger.LogError(ex, 
                "Request failed: {RequestName} ({ElapsedMilliseconds}ms)",
                typeof(TRequest).Name,
                stopwatch.ElapsedMilliseconds);
            throw;
        }
    }
}

Streaming Pipeline Behaviors

For streaming operations, implement IStreamPipelineBehavior<TRequest, TResponse>:

public sealed class StreamLoggingBehavior<TRequest, TResponse>
    : IStreamPipelineBehavior<TRequest, TResponse>
{
    private readonly ILogger<StreamLoggingBehavior<TRequest, TResponse>> _logger;

    public StreamLoggingBehavior(ILogger<StreamLoggingBehavior<TRequest, TResponse>> logger)
    {
        _logger = logger;
    }

    public async IAsyncEnumerable<TResponse> HandleAsync(
        TRequest request,
        StreamHandlerDelegate<TResponse> next,
        [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        var itemCount = 0;
        
        await foreach (var item in next(cancellationToken))
        {
            itemCount++;
            yield return item;
        }
        
        _logger.LogInformation(
            "Stream completed for {RequestName}: {ItemCount} items",
            typeof(TRequest).Name,
            itemCount);
    }
}

🔧 Advanced Usage

Validation Pipeline

public sealed class ValidationBehavior<TRequest, TResponse> 
    : IPipelineBehavior<TRequest, TResponse>
{
    private readonly IEnumerable<IValidator<TRequest>> _validators;

    public ValidationBehavior(IEnumerable<IValidator<TRequest>> validators)
    {
        _validators = validators;
    }

    public async Task<TResponse> HandleAsync(
        TRequest request,
        RequestHandlerDelegate<TResponse> next,
        CancellationToken cancellationToken = default)
    {
        if (_validators.Any())
        {
            var context = new ValidationContext<TRequest>(request);
            
            var validationResults = await Task.WhenAll(
                _validators.Select(v => v.ValidateAsync(context, cancellationToken)));
            
            var failures = validationResults
                .SelectMany(r => r.Errors)
                .Where(f => f != null)
                .ToList();

            if (failures.Any())
            {
                throw new ValidationException(failures);
            }
        }
        
        return await next(cancellationToken);
    }
}

Transaction Pipeline

public sealed class TransactionBehavior<TRequest, TResponse> 
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : ITransactional
{
    private readonly IDbContext _dbContext;

    public TransactionBehavior(IDbContext dbContext)
    {
        _dbContext = dbContext;
    }

    public async Task<TResponse> HandleAsync(
        TRequest request,
        RequestHandlerDelegate<TResponse> next,
        CancellationToken cancellationToken = default)
    {
        using var transaction = await _dbContext.BeginTransactionAsync(cancellationToken);
        
        try
        {
            var response = await next(cancellationToken);
            await transaction.CommitAsync(cancellationToken);
            return response;
        }
        catch
        {
            await transaction.RollbackAsync(cancellationToken);
            throw;
        }
    }
}

⚡ Source Generator

The optional Moderator.Generator package provides compile-time handler discovery and automatic registration:

Automatic Registration

var builder = WebApplication.CreateBuilder(args);

// Register all discovered handlers automatically
builder.Services.AddModerator(config =>
{
    config.DiscoveredTypes = DiscoveredTypes.All;
});

var app = builder.Build();
app.Run();

Generated Discovery Code

The source generator creates a DiscoveredTypes class containing all discovered handlers:

// Auto-generated by Moderator.Generator
internal static partial class DiscoveredTypes
{
    public static IEnumerable<(Type ServiceType, Type ImplementationType)> All =
    [
        // Request Handlers
        (typeof(IRequestHandler<CreateNoteRequest, CreateNoteResponse>), 
         typeof(CreateNoteHandler)),
        
        // Notification Handlers (multiple handlers per notification)
        (typeof(INotificationHandler<NoteCreatedEvent>), 
         typeof(NoteCreatedEventHandler1)),
        (typeof(INotificationHandler<NoteCreatedEvent>), 
         typeof(NoteCreatedEventHandler2)),
        
        // Pipeline Behaviors
        (typeof(IPipelineBehavior<,>), 
         typeof(LoggingBehavior<,>)),
        (typeof(IPipelineBehavior<,>), 
         typeof(ValidationBehavior<,>))
    ];
}

Type-Inferred Extension Methods

The generator also creates extension methods that eliminate the need to specify generic type parameters:

// Auto-generated extension methods
internal static class ModeratorExtensions
{
    public static Task<CreateNoteResponse> SendAsync(
        this IModerator moderator, 
        CreateNoteRequest request, 
        CancellationToken cancellationToken = default)
    {
        return moderator.SendAsync<CreateNoteRequest, CreateNoteResponse>(
            request, 
            cancellationToken);
    }
}

// Usage - no need to specify types!
var response = await moderator.SendAsync(new CreateNoteRequest(title));
// Instead of:
// var response = await moderator.SendAsync<CreateNoteRequest, CreateNoteResponse>(new CreateNoteRequest(title));

Benefits

• Zero Runtime Reflection - All handlers discovered at compile time
• AOT Compatible - Full Native AOT compilation support
• Type Safety - Compile-time errors for missing handlers
• Performance - No assembly scanning at startup
• Cleaner Code - No need to specify generic type parameters
• Auto-Registration - Single line to register all handlers

📊 Performance

Moderator is designed for high-performance scenarios:

• Minimal allocations through struct usage where appropriate
• Async/await throughout with ConfigureAwait(false)
• Efficient service resolution caching
• Zero reflection when using source generator

🤝 Contributing

We welcome contributions! Please see our Contributing Guidelines for details.

Development

# Clone the repository
git clone https://github.com/yourusername/moderator.git

# Build the solution
dotnet build

# Run tests
dotnet test

# Pack NuGet packages
dotnet pack

Reporting Issues

Found a bug or have a feature request? Please open an issue with:

• Clear description of the problem
• Steps to reproduce
• Expected vs actual behavior
• Environment details (.NET version, OS, etc.)

📄 License

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

🔗 Links

• NuGet Package
• API Documentation
• Samples
• Changelog