Indiko.Blocks.Mediation.Abstractions 2.1.1

Indiko.Blocks.Mediation.Abstractions

Core abstractions for implementing CQRS (Command Query Responsibility Segregation) and Mediator patterns in the Indiko framework.

Overview

This package provides the fundamental contracts for building applications using the Mediator pattern and CQRS architecture, enabling clean separation between queries and commands with support for pipeline behaviors and notifications.

Features

• IMediator Interface: Central mediator for request/response handling
• CQRS Support: Separate interfaces for Commands and Queries
• ICommand<TResult>: Command pattern abstraction
• IQuery<TResult>: Query pattern abstraction
• INotification: Pub/sub notification pattern
• Pipeline Behaviors: Cross-cutting concerns (logging, validation, caching)
• Request/Response Pattern: Type-safe request handling
• Saga Support: Long-running transaction patterns
• Async/Await: Full asynchronous support

Installation

dotnet add package Indiko.Blocks.Mediation.Abstractions

Key Interfaces

IMediator

Central interface for sending requests and publishing notifications.

public interface IMediator
{
    // Send a request and get a response
    Task<TResponse> Send<TRequest, TResponse>(TRequest request, CancellationToken cancellationToken = default)
        where TRequest : IRequest<TResponse>;
    
    // Send a command (returns bool)
    Task Send<TRequest>(TRequest request, CancellationToken cancellationToken = default)
        where TRequest : IRequest;
    
    // Publish a notification to multiple handlers
    Task Publish<TNotification>(TNotification notification, CancellationToken cancellationToken = default)
        where TNotification : INotification;
}

ICommand<TResult>

Marker interface for commands (write operations).

// Command that returns a specific result
public interface ICommand<out TResult> : IRequest<TResult>
{
}

// Command that returns bool (success/failure)
public interface ICommand : ICommand<bool>
{
}

IQuery<TResult>

Marker interface for queries (read operations).

// Query that returns a specific result
public interface IQuery<out TResult> : IRequest<TResult>
{
}

// Query that returns a collection
public interface IQuery : IQuery<IEnumerable<IResult>>
{
}

IRequestHandler<TRequest, TResponse>

Handler interface for processing requests.

public interface IRequestHandler<in TRequest, TResponse>
    where TRequest : IRequest<TResponse>
{
    Task<TResponse> Handle(TRequest request, CancellationToken cancellationToken);
}

INotification & INotificationHandler

Pub/sub notification pattern.

public interface INotification
{
    // Marker interface
}

public interface INotificationHandler<in TNotification>
    where TNotification : INotification
{
    Task Handle(TNotification notification, CancellationToken cancellationToken);
}

IPipelineBehavior<TRequest, TResponse>

Cross-cutting concern pipeline.

public interface IPipelineBehavior<in TRequest, TResponse>
    where TRequest : IRequest<TResponse>
{
    Task<TResponse> Handle(
        TRequest request, 
        RequestHandlerDelegate<TResponse> next, 
        CancellationToken cancellationToken);
}

CQRS Pattern

Commands (Write Operations)

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

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

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

    public async Task<Guid> Handle(CreateUserCommand request, CancellationToken cancellationToken)
    {
        var user = new User
        {
            Id = Guid.NewGuid(),
            FirstName = request.FirstName,
            LastName = request.LastName,
            Email = request.Email
        };
        
        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 (Read Operations)

// Define a query
public class GetUserByIdQuery : IQuery<UserDto>
{
    public Guid UserId { get; set; }
}

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

    public GetUserByIdQueryHandler(IUserRepository userRepository)
    {
        _userRepository = userRepository;
    }

    public async Task<UserDto> Handle(GetUserByIdQuery request, CancellationToken cancellationToken)
    {
        var user = await _userRepository.ReadByIdAsync(request.UserId, cancellationToken);
        
        if (user == null)
            return null;
        
        return new UserDto
        {
            Id = user.Id,
            FullName = $"{user.FirstName} {user.LastName}",
            Email = user.Email
        };
    }
}

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

Paged Queries

public class GetUsersPagedQuery : IQuery<PagedList<UserDto>>
{
    public int PageNumber { get; set; } = 1;
    public int PageSize { get; set; } = 20;
    public string SearchTerm { get; set; }
}

public class GetUsersPagedQueryHandler : IRequestHandler<GetUsersPagedQuery, PagedList<UserDto>>
{
    private readonly IUserRepository _userRepository;

    public async Task<PagedList<UserDto>> Handle(GetUsersPagedQuery request, CancellationToken cancellationToken)
    {
        var users = await _userRepository.ReadManyByQueryPagedAsync(
            where: u => string.IsNullOrEmpty(request.SearchTerm) || 
                       u.FirstName.Contains(request.SearchTerm) || 
                       u.LastName.Contains(request.SearchTerm),
            page: request.PageNumber,
            pageSize: request.PageSize,
            cancellationToken: cancellationToken
        );
        
        return new PagedList<UserDto>(
            users.Select(u => new UserDto { /* map properties */ }),
            users.TotalCount,
            users.CurrentPage,
            users.PageSize
        );
    }
}

Notifications (Pub/Sub)

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

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

    public async Task Handle(UserCreatedNotification notification, CancellationToken cancellationToken)
    {
        await _emailService.SendWelcomeEmailAsync(notification.Email);
    }
}

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

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

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

Pipeline Behaviors

Validation Behavior

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())
        {
            var context = new ValidationContext<TRequest>(request);
            var failures = _validators
                .Select(v => v.Validate(context))
                .SelectMany(result => result.Errors)
                .Where(f => f != null)
                .ToList();

            if (failures.Count != 0)
            {
                throw new ValidationException(failures);
            }
        }

        return await next();
    }
}

Logging Behavior

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

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

    public async Task<TResponse> Handle(
        TRequest request, 
        RequestHandlerDelegate<TResponse> next, 
        CancellationToken cancellationToken)
    {
        var requestName = typeof(TRequest).Name;
        
        _logger.LogInformation($"Handling {requestName}");
        var stopwatch = Stopwatch.StartNew();
        
        try
        {
            var response = await next();
            
            stopwatch.Stop();
            _logger.LogInformation(
                $"Handled {requestName} in {stopwatch.ElapsedMilliseconds}ms");
            
            return response;
        }
        catch (Exception ex)
        {
            stopwatch.Stop();
            _logger.LogError(ex, 
                $"Error handling {requestName} after {stopwatch.ElapsedMilliseconds}ms");
            throw;
        }
    }
}

Caching Behavior

public class CachingBehavior<TRequest, TResponse> : IPipelineBehavior<TRequest, TResponse>
    where TRequest : IRequest<TResponse>, ICacheableQuery
{
    private readonly IDistributedCache _cache;

    public async Task<TResponse> Handle(
        TRequest request, 
        RequestHandlerDelegate<TResponse> next, 
        CancellationToken cancellationToken)
    {
        var cacheKey = request.GetCacheKey();
        var cachedResponse = await _cache.GetStringAsync(cacheKey, cancellationToken);
        
        if (cachedResponse != null)
        {
            return JsonSerializer.Deserialize<TResponse>(cachedResponse);
        }
        
        var response = await next();
        
        await _cache.SetStringAsync(
            cacheKey, 
            JsonSerializer.Serialize(response),
            new DistributedCacheEntryOptions
            {
                AbsoluteExpirationRelativeToNow = TimeSpan.FromMinutes(5)
            },
            cancellationToken);
        
        return response;
    }
}

Saga Pattern

public interface ISaga
{
    Task ExecuteAsync(CancellationToken cancellationToken = default);
    Task CompensateAsync(CancellationToken cancellationToken = default);
}

// Example: Order Processing Saga
public class OrderProcessingSaga : ISaga
{
    private readonly IMediator _mediator;

    public OrderProcessingSaga(IMediator mediator)
    {
        _mediator = mediator;
    }

    public async Task ExecuteAsync(CancellationToken cancellationToken = default)
    {
        // Step 1: Reserve inventory
        await _mediator.Send(new ReserveInventoryCommand { ... }, cancellationToken);
        
        // Step 2: Process payment
        await _mediator.Send(new ProcessPaymentCommand { ... }, cancellationToken);
        
        // Step 3: Create shipment
        await _mediator.Send(new CreateShipmentCommand { ... }, cancellationToken);
    }

    public async Task CompensateAsync(CancellationToken cancellationToken = default)
    {
        // Rollback in reverse order
        await _mediator.Send(new CancelShipmentCommand { ... }, cancellationToken);
        await _mediator.Send(new RefundPaymentCommand { ... }, cancellationToken);
        await _mediator.Send(new ReleaseInventoryCommand { ... }, cancellationToken);
    }
}

Service Registration

public class Startup : WebStartup
{
    public override void ConfigureServices(IServiceCollection services)
    {
        base.ConfigureServices(services);
        
        // Register mediator implementation
        services.AddMediator(options =>
        {
            // Scan assemblies for handlers
            options.RegisterServicesFromAssembly(typeof(Startup).Assembly);
            
            // Register pipeline behaviors
            options.AddBehavior<ValidationBehavior<,>>();
            options.AddBehavior<LoggingBehavior<,>>();
            options.AddBehavior<CachingBehavior<,>>();
        });
    }
}

Best Practices

1. Single Responsibility: One handler per request type
2. Immutable Requests: Use read-only properties
3. Thin Controllers: Keep controllers thin, move logic to handlers
4. Query Optimization: Use projections, avoid N+1 queries
5. Command Validation: Validate commands before processing
6. Error Handling: Use pipeline behaviors for consistent error handling
7. Naming Conventions:
   • Commands: CreateUserCommand, UpdateOrderCommand
   • Queries: GetUserByIdQuery, GetOrdersPagedQuery
   • Handlers: CreateUserCommandHandler, GetUserByIdQueryHandler

Benefits

• Decoupling: Loose coupling between sender and receiver
• Single Responsibility: Each handler has one responsibility
• Testability: Easy to unit test handlers in isolation
• Cross-Cutting Concerns: Pipeline behaviors for logging, validation, caching
• CQRS: Clear separation between reads and writes
• Scalability: Query and command handlers can be scaled independently

Target Framework

• .NET 10

Dependencies

• Indiko.Blocks.Common.Abstractions

License

See LICENSE file in the repository root.

Related Packages

• Indiko.Blocks.Mediation.Mediator - MediatR-based implementation
• Indiko.Blocks.Mediation.SimpleMediator - Lightweight implementation
• Indiko.Blocks.EventBus.Abstractions - Event-driven architecture
• Indiko.Common.Abstractions - Common models and interfaces