Indiko.Blocks.Mediation.Mediator 2.2.17

Indiko.Blocks.Mediation.Mediator

MediatR-based implementation of the Indiko mediation abstractions, providing a battle-tested mediator pattern with rich pipeline support.

Overview

This package provides a production-ready implementation of the mediator pattern using MediatR, one of the most popular and well-tested mediator libraries for .NET.

Features

• MediatR Integration: Built on top of MediatR library
• Request/Response: Type-safe request handling
• CQRS Support: Commands and queries with dedicated interfaces
• Pipeline Behaviors: Extensible pipeline for cross-cutting concerns
• Notifications: Pub/sub pattern for domain events
• Streaming: Support for streaming responses
• Polymorphic Dispatch: Handle base types and derived types
• Generic Constraints: Type-safe generic handling
• Dependency Injection: First-class DI support

Installation

dotnet add package Indiko.Blocks.Mediation.Mediator

Quick Start

Configure Services

using Indiko.Blocks.Mediation.Mediator;

public class Startup : WebStartup
{
    public override void ConfigureServices(IServiceCollection services)
    {
        base.ConfigureServices(services);
        
        // Register MediatR-based mediator
        services.AddMediatRMediation(options =>
        {
            // Scan assemblies for handlers
            options.RegisterServicesFromAssembly(typeof(Startup).Assembly);
            
            // Register domain assemblies
            options.RegisterServicesFromAssembly(typeof(CreateUserCommand).Assembly);
        });
        
        // Register pipeline behaviors
        services.AddTransient(typeof(IPipelineBehavior<,>), typeof(LoggingBehavior<,>));
        services.AddTransient(typeof(IPipelineBehavior<,>), typeof(ValidationBehavior<,>));
    }
}

Usage Examples

Commands

// Define command
public class CreateUserCommand : ICommand<Guid>
{
    public string FirstName { get; set; }
    public string LastName { get; set; }
    public string Email { get; set; }
    public string Password { get; set; }
}

// Implement handler
public class CreateUserCommandHandler : IRequestHandler<CreateUserCommand, Guid>
{
    private readonly IUserRepository _userRepository;
    private readonly IPasswordHasher _passwordHasher;
    private readonly IUnitOfWork _unitOfWork;

    public CreateUserCommandHandler(
        IUserRepository userRepository,
        IPasswordHasher passwordHasher,
        IUnitOfWork unitOfWork)
    {
        _userRepository = userRepository;
        _passwordHasher = passwordHasher;
        _unitOfWork = unitOfWork;
    }

    public async Task<Guid> Handle(CreateUserCommand request, CancellationToken cancellationToken)
    {
        // Validate email uniqueness
        var exists = await _userRepository.ExistsAsync(u => u.Email == request.Email, cancellationToken);
        if (exists)
            throw new DomainException("Email already exists");
        
        // Create user
        var user = new User
        {
            Id = Guid.NewGuid(),
            FirstName = request.FirstName,
            LastName = request.LastName,
            Email = request.Email,
            PasswordHash = _passwordHasher.HashPassword(request.Password),
            CreatedAt = DateTime.UtcNow
        };
        
        await _userRepository.AddAsync(user, cancellationToken);
        await _unitOfWork.SaveChangesAsync(cancellationToken);
        
        return user.Id;
    }
}

// Use in controller
[HttpPost]
public async Task<IActionResult> CreateUser([FromBody] CreateUserCommand command)
{
    var userId = await _mediator.Send<CreateUserCommand, Guid>(command);
    return CreatedAtAction(nameof(GetUser), new { id = userId }, userId);
}

Queries

// Define query
public class GetUserByIdQuery : IQuery<UserDto>
{
    public Guid UserId { get; set; }
    
    public GetUserByIdQuery(Guid userId)
    {
        UserId = userId;
    }
}

// Implement handler
public class GetUserByIdQueryHandler : IRequestHandler<GetUserByIdQuery, UserDto>
{
    private readonly IUserRepository _userRepository;
    private readonly IMapper _mapper;

    public GetUserByIdQueryHandler(IUserRepository userRepository, IMapper mapper)
    {
        _userRepository = userRepository;
        _mapper = mapper;
    }

    public async Task<UserDto> Handle(GetUserByIdQuery request, CancellationToken cancellationToken)
    {
        var user = await _userRepository.ReadByIdAsync(
            request.UserId, 
            asNotracking: true, 
            cancellationToken);
        
        return user != null ? _mapper.Map<UserDto>(user) : null;
    }
}

// Use in controller
[HttpGet("{id}")]
public async Task<IActionResult> GetUser(Guid id)
{
    var query = new GetUserByIdQuery(id);
    var user = await _mediator.Send<GetUserByIdQuery, UserDto>(query);
    
    return user != null ? Ok(user) : NotFound();
}

Complex Queries with Includes

public class GetOrderWithDetailsQuery : IQuery<OrderDetailDto>
{
    public Guid OrderId { get; set; }
}

public class GetOrderWithDetailsQueryHandler : IRequestHandler<GetOrderWithDetailsQuery, OrderDetailDto>
{
    private readonly IRepository<Order, Guid> _orderRepository;

    public async Task<OrderDetailDto> Handle(GetOrderWithDetailsQuery request, CancellationToken cancellationToken)
    {
        var order = await _orderRepository.ReadByIdAsync(
            request.OrderId,
            asNotracking: true,
            cancellationToken: cancellationToken,
            includes: o => o.Items, 
                     o => o.Items.Select(i => i.Product),
                     o => o.Customer);
        
        if (order == null)
            return null;
        
        return new OrderDetailDto
        {
            OrderId = order.Id,
            OrderNumber = order.OrderNumber,
            CustomerName = order.Customer.FullName,
            Items = order.Items.Select(i => new OrderItemDto
            {
                ProductName = i.Product.Name,
                Quantity = i.Quantity,
                Price = i.Price
            }).ToList(),
            TotalAmount = order.TotalAmount
        };
    }
}

Notifications (Domain Events)

// Define notification
public class UserCreatedNotification : INotification
{
    public Guid UserId { get; set; }
    public string Email { get; set; }
    public DateTime CreatedAt { get; set; }
}

// Multiple handlers
public class SendWelcomeEmailHandler : INotificationHandler<UserCreatedNotification>
{
    private readonly IEmailService _emailService;
    private readonly ILogger<SendWelcomeEmailHandler> _logger;

    public async Task Handle(UserCreatedNotification notification, CancellationToken cancellationToken)
    {
        _logger.LogInformation($"Sending welcome email to {notification.Email}");
        
        await _emailService.SendAsync(new EmailMessage
        {
            To = notification.Email,
            Subject = "Welcome!",
            Body = "Welcome to our platform!"
        });
    }
}

public class CreateUserProfileHandler : INotificationHandler<UserCreatedNotification>
{
    private readonly IProfileService _profileService;

    public async Task Handle(UserCreatedNotification notification, CancellationToken cancellationToken)
    {
        await _profileService.CreateDefaultProfileAsync(notification.UserId);
    }
}

public class LogUserCreationHandler : INotificationHandler<UserCreatedNotification>
{
    private readonly IAuditService _auditService;

    public async Task Handle(UserCreatedNotification notification, CancellationToken cancellationToken)
    {
        await _auditService.LogAsync(new AuditEntry
        {
            Action = "UserCreated",
            UserId = notification.UserId,
            Timestamp = notification.CreatedAt
        });
    }
}

// Publish notification
await _mediator.Publish(new UserCreatedNotification
{
    UserId = user.Id,
    Email = user.Email,
    CreatedAt = user.CreatedAt
});

Pipeline Behaviors

Validation with FluentValidation

public class CreateUserCommandValidator : AbstractValidator<CreateUserCommand>
{
    public CreateUserCommandValidator()
    {
        RuleFor(x => x.FirstName)
            .NotEmpty()
            .MaximumLength(100);
            
        RuleFor(x => x.LastName)
            .NotEmpty()
            .MaximumLength(100);
            
        RuleFor(x => x.Email)
            .NotEmpty()
            .EmailAddress();
            
        RuleFor(x => x.Password)
            .NotEmpty()
            .MinimumLength(8)
            .Matches(@"[A-Z]").WithMessage("Password must contain uppercase")
            .Matches(@"[a-z]").WithMessage("Password must contain lowercase")
            .Matches(@"\d").WithMessage("Password must contain digit");
    }
}

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

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

    public async Task<TResponse> Handle(
        TRequest request, 
        RequestHandlerDelegate<TResponse> next, 
        CancellationToken cancellationToken)
    {
        if (!_validators.Any())
            return await next();

        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();
    }
}

Performance Logging

public class PerformanceBehavior<TRequest, TResponse> : IPipelineBehavior<TRequest, TResponse>
    where TRequest : IRequest<TResponse>
{
    private readonly ILogger<PerformanceBehavior<TRequest, TResponse>> _logger;

    public async Task<TResponse> Handle(
        TRequest request, 
        RequestHandlerDelegate<TResponse> next, 
        CancellationToken cancellationToken)
    {
        var requestName = typeof(TRequest).Name;
        var stopwatch = Stopwatch.StartNew();
        
        try
        {
            var response = await next();
            
            stopwatch.Stop();
            var elapsedMilliseconds = stopwatch.ElapsedMilliseconds;
            
            if (elapsedMilliseconds > 500) // Slow query threshold
            {
                _logger.LogWarning(
                    "Long Running Request: {RequestName} ({ElapsedMilliseconds} ms) {@Request}",
                    requestName, 
                    elapsedMilliseconds, 
                    request);
            }
            else
            {
                _logger.LogInformation(
                    "Request: {RequestName} ({ElapsedMilliseconds} ms)",
                    requestName, 
                    elapsedMilliseconds);
            }
            
            return response;
        }
        catch (Exception ex)
        {
            stopwatch.Stop();
            _logger.LogError(ex, 
                "Request Failed: {RequestName} ({ElapsedMilliseconds} ms)",
                requestName, 
                stopwatch.ElapsedMilliseconds);
            throw;
        }
    }
}

Transaction Behavior

public class TransactionBehavior<TRequest, TResponse> : IPipelineBehavior<TRequest, TResponse>
    where TRequest : IRequest<TResponse>
{
    private readonly IUnitOfWork _unitOfWork;
    private readonly ILogger<TransactionBehavior<TRequest, TResponse>> _logger;

    public async Task<TResponse> Handle(
        TRequest request, 
        RequestHandlerDelegate<TResponse> next, 
        CancellationToken cancellationToken)
    {
        // Only use transactions for commands (not queries)
        if (request is IQuery<TResponse>)
            return await next();

        try
        {
            await _unitOfWork.BeginTransactionAsync(cancellationToken);
            
            var response = await next();
            
            await _unitOfWork.CommitTransactionAsync(cancellationToken);
            
            return response;
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Transaction failed for {RequestName}", typeof(TRequest).Name);
            await _unitOfWork.RollbackTransactionAsync(cancellationToken);
            throw;
        }
    }
}

Streaming Responses

// Streaming query
public class StreamUsersQuery : IStreamRequest<UserDto>
{
    public string SearchTerm { get; set; }
}

// Streaming handler
public class StreamUsersQueryHandler : IStreamRequestHandler<StreamUsersQuery, UserDto>
{
    private readonly IUserRepository _userRepository;

    public async IAsyncEnumerable<UserDto> Handle(
        StreamUsersQuery request, 
        [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        var users = await _userRepository.AsEnumerableAsync(cancellationToken);
        
        await foreach (var user in users.WithCancellation(cancellationToken))
        {
            if (string.IsNullOrEmpty(request.SearchTerm) || 
                user.FullName.Contains(request.SearchTerm, StringComparison.OrdinalIgnoreCase))
            {
                yield return new UserDto
                {
                    Id = user.Id,
                    FullName = user.FullName,
                    Email = user.Email
                };
            }
        }
    }
}

// Use streaming
await foreach (var user in _mediator.CreateStream(new StreamUsersQuery { SearchTerm = "John" }))
{
    Console.WriteLine($"{user.FullName} - {user.Email}");
}

Advanced Registration

services.AddMediatRMediation(options =>
{
    // Register from multiple assemblies
    options.RegisterServicesFromAssemblies(
        typeof(Startup).Assembly,
        typeof(CreateUserCommand).Assembly,
        typeof(GetUserByIdQuery).Assembly
    );
    
    // Custom lifetime
    options.Lifetime = ServiceLifetime.Scoped;
    
    // Register open generics
    options.AddOpenBehavior(typeof(LoggingBehavior<,>));
    options.AddOpenBehavior(typeof(ValidationBehavior<,>));
    options.AddOpenBehavior(typeof(TransactionBehavior<,>));
});

Testing

Unit Testing Handlers

[Fact]
public async Task CreateUserCommand_Should_CreateUser()
{
    // Arrange
    var userRepoMock = new Mock<IUserRepository>();
    var unitOfWorkMock = new Mock<IUnitOfWork>();
    var passwordHasherMock = new Mock<IPasswordHasher>();
    
    passwordHasherMock
        .Setup(x => x.HashPassword(It.IsAny<string>()))
        .Returns("hashed_password");
    
    var handler = new CreateUserCommandHandler(
        userRepoMock.Object,
        passwordHasherMock.Object,
        unitOfWorkMock.Object
    );
    
    var command = new CreateUserCommand
    {
        FirstName = "John",
        LastName = "Doe",
        Email = "john@example.com",
        Password = "Password123"
    };
    
    // Act
    var userId = await handler.Handle(command, CancellationToken.None);
    
    // Assert
    Assert.NotEqual(Guid.Empty, userId);
    userRepoMock.Verify(x => x.AddAsync(It.IsAny<User>(), It.IsAny<CancellationToken>()), Times.Once);
    unitOfWorkMock.Verify(x => x.SaveChangesAsync(It.IsAny<CancellationToken>()), Times.Once);
}

Target Framework

• .NET 10

Dependencies

• Indiko.Blocks.Mediation.Abstractions
• MediatR (13.0+)
• MediatR.Extensions.Microsoft.DependencyInjection

License

See LICENSE file in the repository root.

Related Packages

• Indiko.Blocks.Mediation.Abstractions - Core mediation abstractions
• Indiko.Blocks.Mediation.SimpleMediator - Lightweight implementation
• Indiko.Blocks.EventBus.Abstractions - Event-driven architecture
• FluentValidation - Validation library (optional)