👁 Image
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)