M3diator 2.1.0

M3diator

A lightweight, high-performance Mediator pattern implementation in .NET, inspired by MediatR.

What is the Mediator Pattern?

The Mediator pattern is a behavioral design pattern that promotes loose coupling between components by having them interact indirectly through a central mediator object, instead of directly with each other.

Benefits:

• Reduced Coupling: Objects don't need direct references to each other. They only know the mediator. This aligns with SOLID principles like Dependency Inversion (DIP) and makes the system easier to modify.
• Improved Maintainability: Changes within one component are less likely to impact others. Adding new components or handlers often doesn't require changing existing ones (Open/Closed Principle - OCP).
• Single Responsibility Principle (SRP): Each handler focuses on a single request or notification type. The mediator focuses solely on dispatching.
• Simplified Communication: Centralizes complex communication logic that would otherwise be scattered among multiple components.

In the context of CQRS (Command Query Responsibility Segregation), the Mediator pattern is often used to decouple the sender of a command or query from its handler.

What is M3diator?

M3diator is a .NET library that provides an elegant and efficient way to implement the Mediator pattern. It helps you build cleaner, more maintainable applications by decoupling in-process message sending and handling.

It draws heavy inspiration from the popular MediatR library but aims for simplicity and optimal performance, leveraging modern .NET features.

Installation

Install M3diator via the NuGet Package Manager console or by searching for M3diator in the NuGet Package Manager UI.

Install-Package M3diator
Install-Package M3diator.Abstractions

(Note: M3diator depends on M3diator.Abstractions, so installing M3diator usually suffices. Install M3diator.Abstractions separately if you only need the contracts in a specific project).

Configuration

Register M3diator and its handlers in your application's service collection, typically in Program.cs (for .NET 6+) or Startup.cs.

// Program.cs (.NET 6+)

using M3diator.Extensions; // Add this using statement

var builder = WebApplication.CreateBuilder(args);

// Add M3diator and scan assemblies for handlers
// Replace 'typeof(Program).Assembly' with the assembly(ies) containing your handlers
builder.Services.AddM3diator(cfg =>
{
    cfg.RegisterServicesFromAssembly(typeof(Program).Assembly);
    // Optionally register pipeline behaviors
    // cfg.AddPipelineBehavior(typeof(YourLoggingBehavior<,>));
});

// ... other services

var app = builder.Build();

// ... configure middleware

app.Run();

The AddM3diator extension method registers the core IMediator, ISender, and IPublisher interfaces. RegisterServicesFromAssembly scans the specified assembly (or assemblies) for implementations of IRequestHandler<,>, IRequestHandler<>, and INotificationHandler<> and registers them with the dependency injection container.

Usage

M3diator revolves around three main message types: Requests (Commands/Queries) and Notifications (Events).

1. Requests (Commands and Queries)

Requests represent an action to perform or data to retrieve. They can optionally return a value.

• IRequest<TResponse>: Represents a request that returns a value of type TResponse. Typically used for Queries.
• IRequest: Represents a request that does not return a value (implicitly returns Unit). Typically used for Commands. (M3diator.Abstractions.Unit is a void marker type).

Defining Requests

using M3diator.Abstractions;

/// <summary>
/// Represents a query to retrieve a specific customer's details.
/// Implements IRequest<TResponse> because it returns a CustomerDto.
/// </summary>
public class GetCustomerByIdQuery : IRequest<CustomerDto>
{
    public int CustomerId { get; set; }
}

/// <summary>
/// Represents a command to create a new customer.
/// Implements IRequest (or IRequest<Unit>) because it performs an action
/// without returning specific data (fire-and-forget style).
/// </summary>
public class CreateCustomerCommand : IRequest // Equivalent to IRequest<Unit>
{
    public string Name { get; set; } = string.Empty;
    public string Email { get; set; } = string.Empty;
}

// Example DTO
public class CustomerDto
{
    public int Id { get; set; }
    public string Name { get; set; } = string.Empty;
}

Defining Request Handlers

Handlers contain the logic to process a specific request type. Each request type must have exactly one handler.

• IRequestHandler<TRequest, TResponse>: Handles requests implementing IRequest<TResponse>.
• IRequestHandler<TRequest>: Handles requests implementing IRequest (which implicitly means IRequest<Unit>).

using M3diator.Abstractions;
using System.Threading;
using System.Threading.Tasks;

/// <summary>
/// Handles the GetCustomerByIdQuery request.
/// Dependencies (like a repository) can be injected via the constructor.
/// </summary>
public class GetCustomerByIdQueryHandler : IRequestHandler<GetCustomerByIdQuery, CustomerDto>
{
    // Example dependency (replace with your actual data access)
    private readonly ICustomerRepository _customerRepository;

    public GetCustomerByIdQueryHandler(ICustomerRepository customerRepository)
    {
        _customerRepository = customerRepository;
    }

    /// <summary>
    /// Processes the incoming query.
    /// </summary>
    /// <param name="request">The query object containing the CustomerId.</param>
    /// <param name="cancellationToken">Cancellation token.</param>
    /// <returns>A Task containing the CustomerDto or null if not found.</returns>
    public async Task<CustomerDto> Handle(GetCustomerByIdQuery request, CancellationToken cancellationToken)
    {
        // Simulate fetching data
        await Task.Delay(10, cancellationToken); // Simulate async work

        if (request.CustomerId == 1)
        {
            return new CustomerDto { Id = request.CustomerId, Name = "John Doe" };
        }
        return null; // Or throw an exception if preferred
    }
}

/// <summary>
/// Handles the CreateCustomerCommand request.
/// </summary>
public class CreateCustomerCommandHandler : IRequestHandler<CreateCustomerCommand> // No TResponse needed
{
    private readonly ICustomerRepository _customerRepository;

    public CreateCustomerCommandHandler(ICustomerRepository customerRepository)
    {
         _customerRepository = customerRepository;
    }

    /// <summary>
    /// Processes the incoming command.
    /// </summary>
    /// <param name="request">The command object containing customer details.</param>
    /// <param name="cancellationToken">Cancellation token.</param>
    /// <returns>A Task containing Unit (representing void).</returns>
    public async Task<Unit> Handle(CreateCustomerCommand request, CancellationToken cancellationToken)
    {
        // Simulate saving data
        Console.WriteLine($"Creating customer: {request.Name}, Email: {request.Email}");
        await Task.Delay(20, cancellationToken); // Simulate async work
        // _customerRepository.Add(new Customer { Name = request.Name, Email = request.Email });
        // await _customerRepository.SaveChangesAsync(cancellationToken);

        return Unit.Value; // Return Unit.Value for void handlers
    }
}

// Dummy repository interface for example
public interface ICustomerRepository { /* ... methods ... */ }

Sending Requests

Inject IMediator or ISender into your classes (e.g., Controllers, Services) and use the Send method.

using M3diator.Abstractions;
using Microsoft.AspNetCore.Mvc;
using System.Threading.Tasks;

[ApiController]
[Route("[controller]")]
public class CustomersController : ControllerBase
{
    private readonly IMediator _mediator; // Or ISender

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

    /// <summary>
    /// GET endpoint to fetch a customer by ID.
    /// </summary>
    [HttpGet("{id}")]
    public async Task<IActionResult> GetCustomer(int id)
    {
        var query = new GetCustomerByIdQuery { CustomerId = id };
        CustomerDto? customer = await _mediator.Send(query); // Use Send for requests

        if (customer == null)
        {
            return NotFound();
        }
        return Ok(customer);
    }

    /// <summary>
    /// POST endpoint to create a new customer.
    /// </summary>
    [HttpPost]
    public async Task<IActionResult> CreateCustomer([FromBody] CreateCustomerCommand command)
    {
        await _mediator.Send(command); // Send returns Task<Unit> which can be awaited
        // Usually return CreatedAtAction or Ok
        return Ok();
    }
}

2. Notifications (Events)

Notifications represent something that has happened (an event). They are dispatched to multiple handlers.

• INotification: Marker interface for notification messages.

Defining Notifications

using M3diator.Abstractions;

/// <summary>
/// Notification published after a customer has been successfully created.
/// </summary>
public class CustomerCreatedNotification : INotification
{
    public int CustomerId { get; }
    public string CustomerName { get; }

    public CustomerCreatedNotification(int customerId, string customerName)
    {
        CustomerId = customerId;
        CustomerName = customerName ?? string.Empty;
    }
}

Defining Notification Handlers

A single notification can have zero or many handlers. Handlers implement INotificationHandler<TNotification>.

using M3diator.Abstractions;
using System.Threading;
using System.Threading.Tasks;

/// <summary>
/// Handles the CustomerCreatedNotification to send a welcome email.
/// </summary>
public class SendWelcomeEmailHandler : INotificationHandler<CustomerCreatedNotification>
{
    /// <summary>
    /// Handles the notification.
    /// </summary>
    /// <param name="notification">The notification object.</param>
    /// <param name="cancellationToken">Cancellation token.</param>
    /// <returns>A Task representing the completion of handling.</returns>
    public async Task Handle(CustomerCreatedNotification notification, CancellationToken cancellationToken)
    {
        Console.WriteLine($"Handler 1: Sending welcome email to customer {notification.CustomerName} (ID: {notification.CustomerId})");
        
        await Task.Delay(15, cancellationToken); // Simulate I/O
    }
}

//// <summary>
/// Another handler for the same CustomerCreatedNotification, perhaps for logging.
/// </summary>
public class LogCustomerCreationHandler : INotificationHandler<CustomerCreatedNotification>
{
    /// <summary>
    /// Handles the notification.
    /// </summary>
    /// <param name="notification">The notification object.</param>
    /// <param name="cancellationToken">Cancellation token.</param>
    /// <returns>A Task representing the completion of handling.</returns>
    public async Task Handle(CustomerCreatedNotification notification, CancellationToken cancellationToken)
    {        
        Console.WriteLine($"Handler 2: Logging creation of customer {notification.CustomerName} (ID: {notification.CustomerId})");
        await Task.CompletedTask;
    }
}

Publishing Notifications

Inject IMediator or IPublisher and use the Publish method. Publish will locate all registered handlers for the notification type and invoke them (typically concurrently, but M3diator awaits their completion by default).

// Inside CreateCustomerCommandHandler.Handle method, after saving the customer:

public async Task<Unit> Handle(CreateCustomerCommand request, CancellationToken cancellationToken)
{
    // ... (code to save customer and get the new customerId/Name)
    var newCustomerId = 123; // Example Id
    var customerName = request.Name;

    Console.WriteLine($"Customer {customerName} created with ID {newCustomerId}.");

    // Publish a notification event
    var notification = new CustomerCreatedNotification(newCustomerId, customerName);
    await _mediator.Publish(notification, cancellationToken); // Use Publish for notifications

    Console.WriteLine("Customer creation notification published.");

    return Unit.Value;
}

// Need to inject IMediator or IPublisher into the handler
public class CreateCustomerCommandHandler : IRequestHandler<CreateCustomerCommand>
{
    private readonly ICustomerRepository _customerRepository;
    private readonly IMediator _mediator; // Inject Mediator/Publisher

    public CreateCustomerCommandHandler(ICustomerRepository customerRepository, IMediator mediator)
    {
         _customerRepository = customerRepository;
         _mediator = mediator; // Assign injected instance
    }
    // ... Handle method as above ...
}

3. Pipeline Behaviors

Pipeline behaviors allow you to wrap inner request handlers with additional logic, enabling cross-cutting concerns like logging, validation, performance monitoring, or transaction management without cluttering your core handler logic.

• IPipelineBehavior<TRequest, TResponse>: Interface to implement for pipeline behaviors.

Defining a Pipeline Behavior

using M3diator.Abstractions;
using System;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;

/// <summary>
/// Example pipeline behavior for logging request execution time.
/// </summary>
/// <typeparam name="TRequest">Type of the request being handled.</typeparam>
/// <typeparam name="TResponse">Type of the response from the handler.</typeparam>
public class PerformanceLoggingBehavior<TRequest, TResponse> : IPipelineBehavior<TRequest, TResponse>
    where TRequest : IRequest<TResponse> // Constrain TRequest to match the interface
{
    /// <summary>
    /// Intercepts the request handling process.
    /// </summary>
    /// <param name="request">The incoming request.</param>
    /// <param name="next">The delegate representing the next action in the pipeline (usually the handler itself or the next behavior).</param>
    /// <param name="cancellationToken">Cancellation token.</param>
    /// <returns>The response from the inner handler.</returns>
    public async Task<TResponse> Handle(TRequest request, RequestHandlerDelegate<TResponse> next, CancellationToken cancellationToken)
    {
        var stopwatch = Stopwatch.StartNew();

        Console.WriteLine($"[Pipeline] Handling request {typeof(TRequest).Name}...");

        try
        {
            // Call the next delegate/handler in the chain
            var response = await next();

            stopwatch.Stop();

            Console.WriteLine($"[Pipeline] Handled {typeof(TRequest).Name} in {stopwatch.ElapsedMilliseconds}ms.");

            return response;
        }
        catch (Exception ex)
        {
            stopwatch.Stop();

            Console.WriteLine($"[Pipeline] Exception handling {typeof(TRequest).Name} after {stopwatch.ElapsedMilliseconds}ms: {ex.Message}");
            throw; // Re-throw the exception to maintain original behavior
        }
    }
}

Registering Pipeline Behaviors

Register pipeline behaviors with the DI container. The order of registration matters; they execute in the order they are added.

// Program.cs or Startup.cs

builder.Services.AddM3diator(cfg =>
{
    cfg.RegisterServicesFromAssembly(typeof(Program).Assembly);

    // Register pipeline behaviors - Order matters!
    // These apply to all requests matching the generic constraints.
    cfg.AddPipelineBehavior(typeof(PerformanceLoggingBehavior<,>));
    // cfg.AddPipelineBehavior(typeof(ValidationBehavior<,>)); // Example: Add validation next
});

// Alternatively, using standard DI registration (useful for more complex scenarios):
// builder.Services.AddScoped(typeof(IPipelineBehavior<,>), typeof(PerformanceLoggingBehavior<,>));
// builder.Services.AddScoped(typeof(IPipelineBehavior<,>), typeof(ValidationBehavior<,>));

When a request is sent using _mediator.Send(), M3diator constructs a pipeline: the request flows through each registered IPipelineBehavior before reaching the actual IRequestHandler, and the response flows back out through the behaviors in reverse order.

4. Streaming Requests

Beyond single request/response and notifications, M3diator supports streaming requests where a single request results in a stream of multiple responses over time. This is useful for scenarios like processing large datasets without loading everything into memory, handling long-running operations that report progress, or returning sequences where items become available incrementally.

• IStreamRequest<TResponse>: Represents a request that will yield a stream of TResponse items.
• IStreamRequestHandler<TRequest, TResponse>: Defines the handler responsible for processing an IStreamRequest<TResponse> and generating the IAsyncEnumerable<TResponse>.

Defining Stream Requests

Define a class or record implementing IStreamRequest<TResponse>, where TResponse is the type of item in the resulting stream.

using M3diator; // Or M3diator.Abstractions if interfaces are there

/// <summary>
/// Represents a request to stream a sequence of data records.
/// </summary>
/// <param name="Filter">Criteria to filter the data.</param>
public record StreamDataRequest(string Filter) : IStreamRequest<DataRecord>;

// Example response item DTO
public record DataRecord(int Id, string Payload);

Defining Stream Request Handlers

Implement IStreamRequestHandler<TRequest, TResponse>. The Handle method must return an IAsyncEnumerable<TResponse>. Use async IAsyncEnumerable and yield return to produce items asynchronously. Crucially, check the CancellationToken frequently within the handler to support cancellation during potentially long-running stream generation.

using M3diator; // Or M3diator.Abstractions
using System.Collections.Generic;
using System.Runtime.CompilerServices; // For [EnumeratorCancellation]
using System.Threading;
using System.Threading.Tasks;

/// <summary>
/// Handles the StreamDataRequest to generate a stream of DataRecord objects.
/// </summary>
public class StreamDataRequestHandler : IStreamRequestHandler<StreamDataRequest, DataRecord>
{
    // Example dependency
    // private readonly IDataSource _dataSource;
    // public StreamDataRequestHandler(IDataSource dataSource) { /* ... */ }

    /// <summary>
    /// Processes the request and generates the asynchronous stream.
    /// </summary>
    /// <param name="request">The stream request.</param>
    /// <param name="cancellationToken">Cancellation token.</param>
    /// <returns>An asynchronous stream of DataRecord items.</returns>
    public async IAsyncEnumerable<DataRecord> Handle(
        StreamDataRequest request,
        [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        Console.WriteLine(<span class="math-inline">"Streaming data records matching filter: {request.Filter}");
        // Simulate fetching pages or chunks of data
        for (int i = 1; i <= 5; i++) // Example: 5 chunks
        {
        // Check for cancellation before potentially expensive work/IO
        cancellationToken.ThrowIfCancellationRequested();
        Console.WriteLine(</span>"Fetching data chunk {i}...");
            await Task.Delay(100, cancellationToken); // Simulate async I/O

            // Yield items for the current chunk
            for (int j = 0; j < 3; j++) // Example: 3 items per chunk
            {
                // Check cancellation before yielding
                cancellationToken.ThrowIfCancellationRequested();

                int recordId = (i - 1) * 3 + j + 1;
                yield return new DataRecord(recordId, $"Record {recordId} for filter '{request.Filter}'");
            }
        }
        Console.WriteLine("Finished streaming data records.");
    }
}

Consuming Streams

Inject IMediator or ISender and use the CreateStream method. Consume the returned IAsyncEnumerable<TResponse> using await foreach.

using M3diator; // Or M3diator.Abstractions
using Microsoft.AspNetCore.Mvc; // Example using ASP.NET Core
using System.Threading.Tasks;
using System.Threading; // For CancellationToken

[ApiController]
[Route("[controller]")]
public class DataStreamController : ControllerBase
{
    private readonly IMediator _mediator;

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

    /// <summary>
    /// GET endpoint to initiate and consume a data stream.
    /// </summary>
    [HttpGet("stream")]
    public async Task StreamData([FromQuery] string filter = "default", CancellationToken cancellationToken = default)
    {
        var request = new StreamDataRequest(filter);

        Console.WriteLine($"Controller: Calling CreateStream for filter '{filter}'...");

        // Use CreateStream to get the async stream
        IAsyncEnumerable<DataRecord> dataStream = _mediator.CreateStream(request, cancellationToken);

        // Consume the stream asynchronously
        // Response headers must be sent before the first await foreach iteration if streaming to HTTP response
        // Consider using IActionResult like `return new StreamResult(dataStream)` in real web scenarios

        int count = 0;
        await foreach (var record in dataStream.WithCancellation(cancellationToken)) // Essential to propagate cancellation
        {
            // Process each item as it arrives
            count++;
            Console.WriteLine(<span class="math-inline">"Controller: Received record {record.Id}: {record.Payload}");
            // Optional: Add delay to simulate processing
            // await Task.Delay(50, cancellationToken);
        }
        Console.WriteLine(</span>"Controller: Finished consuming stream. Total items received: {count}.");

        // In a real API, you might stream this directly to the response body.
        // This example just consumes and logs. You would typically return an appropriate ActionResult.
        // return Ok(); // Or perhaps no explicit return if streaming directly to Response.BodyWriter
    }
}

Performance Benchmarks

Performance is a key consideration for M3diator. Below are the benchmark results comparing M3diator's Send and Publish operations against MediatR.

BenchmarkDotNet v0.14.0, Windows 10 (10.0.19045.5737/22H2/2022Update)
Intel Core i7-4790 CPU 3.60GHz (Haswell), 1 CPU, 8 logical and 4 physical cores
.NET SDK 9.0.203
  [Host]   : .NET 9.0.4 (9.0.425.16305), X64 RyuJIT AVX2
  .NET 9.0 : .NET 9.0.4 (9.0.425.16305), X64 RyuJIT AVX2

Job=.NET 9.0  Runtime=.NET 9.0  

Key Takeaways:

• High Performance: M3diator demonstrates high performance across all operations (Send, Publish, CreateStream) with low overhead, typically executing in the microsecond or even sub-microsecond range for the core dispatch logic.
• Efficient Streaming: The CreateStream operation shows extremely low setup overhead (~75 ns) and scales predictably with the number of items yielded by the handler. It provides an efficient way to handle sequences of data compared to multiple individual requests.
• Low Allocation: M3diator operations exhibit low memory allocation for the dispatching mechanism itself. While not strictly zero (e.g., CreateStream setup allocates ~192 B, Send without pipeline ~832 B), the overhead allocation is minimal, making it suitable for performance-sensitive applications. Allocations primarily scale with the data being processed (handler/item allocations).
• Minimal Pipeline Impact: The pipeline behavior mechanism adds very little overhead per behavior (typically sub-microsecond), allowing for cross-cutting concerns without significant performance degradation.

These results suggest that M3diator is a highly performant choice for applications requiring efficient in-process messaging, including request/response, notifications, and data streaming patterns, where mediator overhead is a critical factor.

Contributing

Contributions are welcome! Please feel free to open an issue or submit a pull request. (Add more details here if needed: contribution guidelines, code style, etc.)

License

This project <img alt="Brasil" src="https://github.com/JohnSalazar/microservices-go-authentication/assets/16736914/3ecb04fb-b2ce-4e8b-b492-99c5c5c4b317" width="20" height="14" /> is licensed under the MIT License.