FS.Mediator 10.0.5

FS.Mediator

A comprehensive, high-performance mediator library for .NET with advanced streaming capabilities and enterprise-grade resilience patterns.

FS.Mediator isn't just another mediator implementation. It's a complete solution for building scalable, resilient applications with sophisticated data processing capabilities. Whether you're building microservices, processing large datasets, or creating real-time applications, FS.Mediator provides the tools you need.

✨ Why FS.Mediator?

Imagine you're building a modern application that needs to handle thousands of requests per second, process large data streams, and remain resilient under pressure. Traditional mediator libraries give you the basic request/response pattern, but when you need enterprise-grade features like circuit breakers, backpressure handling, and streaming operations, you're on your own.

FS.Mediator bridges this gap by providing everything you need in one cohesive package:

• 🎯 Clean Architecture: Decoupled request/response and notification patterns
• 🌊 Advanced Streaming: Process millions of records without memory issues
• 🛡️ Built-in Resilience: Circuit breakers, retry policies, and error handling
• ⚡ Performance Optimized: Backpressure handling and resource management
• 📊 Health Monitoring: Real-time diagnostics and performance tracking
• 🔧 Highly Configurable: Multiple presets for common scenarios

🚀 Quick Start

Installation

dotnet add package FS.Mediator

Basic Setup

// Program.cs
using FS.Mediator.Extensions;

var builder = WebApplication.CreateBuilder(args);

// Add FS.Mediator with automatic handler discovery
builder.Services.AddFSMediator();

var app = builder.Build();

Your First Request

// Define a request
public record GetUserQuery(int Id) : IRequest<User>;

// Create a handler
public class GetUserHandler : IRequestHandler<GetUserQuery, User>
{
    public async Task<User> HandleAsync(GetUserQuery request, CancellationToken cancellationToken)
    {
        // Your business logic here
        return await _userRepository.GetByIdAsync(request.Id);
    }
}

// Use in your controller
[ApiController]
public class UsersController : ControllerBase
{
    private readonly IMediator _mediator;

    public UsersController(IMediator mediator) => _mediator = mediator;

    [HttpGet("{id}")]
    public async Task<User> GetUser(int id)
    {
        return await _mediator.SendAsync(new GetUserQuery(id));
    }
}

// Minimal API
app.MapGet("/users/{id}", async (int id, IMediator mediator) =>
    await mediator.SendAsync(new GetUserQuery(id)));

That's it! You now have a clean, decoupled architecture with powerful features ready to use.

🌟 Key Features

1. Traditional Mediator Pattern

Handle requests, responses, and notifications with clean separation of concerns:

// Request with response
public record CreateOrderCommand(string CustomerName, decimal Amount) : IRequest<Order>;

// Request without response
public record LogUserActivityCommand(int UserId, string Action) : IRequest<Unit>;

// Notification (multiple handlers)
public record OrderCreatedNotification(Order Order) : INotification;

2. Advanced Streaming Operations

Process large datasets efficiently without loading everything into memory:

// Streaming request
public record GetAllUsersStreamQuery(string Department) : IStreamRequest<User>;

// Streaming handler
public class GetAllUsersStreamHandler : IStreamRequestHandler<GetAllUsersStreamQuery, User>
{
    public async IAsyncEnumerable<User> HandleAsync(
        GetAllUsersStreamQuery request, 
        [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        await foreach (var user in _repository.GetUsersByDepartmentAsync(request.Department))
        {
            yield return user; // Memory-efficient streaming
        }
    }
}

// Usage - process millions of records efficiently
await foreach (var user in _mediator.CreateStream(new GetAllUsersStreamQuery("Engineering")))
{
    await ProcessUserAsync(user); // Process each user as it arrives
}

3. Enterprise-Grade Resilience

Built-in patterns to handle failures gracefully:

// Add resilience with simple configuration
builder.Services
    .AddFSMediator()
    .AddRetryBehavior(RetryPreset.Database)           // Intelligent database retry
    .AddCircuitBreakerBehavior(CircuitBreakerPreset.ExternalApi)  // API protection
    .AddLoggingBehavior()                             // Comprehensive logging
    .AddPerformanceBehavior();                        // Performance monitoring

4. Streaming with Resilience

Combine streaming with enterprise patterns for robust data processing:

builder.Services
    .AddFSMediator()
    .AddStreamingResiliencePackage()                  // Complete streaming protection
    .AddStreamingBackpressureBehavior(BackpressurePreset.Analytics)  // Handle load spikes
    .AddStreamingHealthCheckBehavior(HealthCheckPreset.LongRunning); // Monitor health

5. Request Validation

Validate requests before they reach your handlers using IValidator<T>. Validation runs automatically in the pipeline and throws ValidationException when rules fail:

// Define validation rules
public class CreateOrderCommandValidator : IValidator<CreateOrderCommand>
{
    public async Task<ValidationResult> ValidateAsync(CreateOrderCommand request, CancellationToken cancellationToken)
    {
        var errors = new List<ValidationError>();

        if (string.IsNullOrWhiteSpace(request.CustomerName))
            errors.Add(new ValidationError("CustomerName", "Customer name is required.", ValidationSeverity.Error));

        if (request.Amount <= 0)
            errors.Add(new ValidationError("Amount", "Amount must be greater than zero.", ValidationSeverity.Error));

        if (request.Amount > 100_000)
            errors.Add(new ValidationError("Amount", "Amount exceeds recommended limit.", ValidationSeverity.Warning));

        return new ValidationResult(errors);
    }
}

// Register validation behavior
builder.Services
    .AddFSMediator()
    .AddValidationBehavior();  // Auto-discovers all IValidator<T> implementations

Validation severity levels allow fine-grained control:

Severity	Behavior
Error	Throws ValidationException, stops pipeline execution
Warning	Logged but does not stop execution
Info	Informational, captured in validation result

6. Centralized Exception Handling

Handle exceptions consistently across your application with IExceptionHandler<TRequest, TResponse, TException>:

// Handle specific exception types per request
public class CreateOrderExceptionHandler
    : IExceptionHandler<CreateOrderCommand, OrderResult, DbUpdateException>
{
    public async Task<OrderResult> HandleAsync(
        CreateOrderCommand request,
        DbUpdateException exception,
        CancellationToken cancellationToken)
    {
        _logger.LogError(exception, "Failed to create order for customer {CustomerId}", request.CustomerId);

        // Return a fallback response instead of letting the exception propagate
        return new OrderResult(Id: 0, Total: 0, Error: "Order creation failed. Please try again.");
    }
}

// Register exception handling behavior
builder.Services
    .AddFSMediator()
    .AddExceptionHandlingBehavior(options =>
    {
        options.RethrowUnhandledExceptions = true;  // Rethrow exceptions without a matching handler
    });

7. Timeout Management

Prevent requests from running indefinitely with built-in timeout behavior and ready-made presets:

// Use a preset
builder.Services
    .AddFSMediator()
    .AddTimeoutBehavior(TimeoutPreset.Standard);

// Or configure manually
builder.Services
    .AddFSMediator()
    .AddTimeoutBehavior(options =>
    {
        options.DefaultTimeout = TimeSpan.FromSeconds(45);
    });

Choose the right preset for your scenario:

Preset	Timeout	Use Case
QuickResponse	5s	Cache lookups, simple CRUD
Standard	30s	Database queries, API calls
LongRunning	2min	Reports, file processing
Batch	10min	Data migrations, bulk ops
ExternalApi	15s	Third-party API calls

8. Pre/Post Processors

Execute logic before and after every request handler using IRequestPreProcessor<TRequest> and IRequestPostProcessor<TRequest, TResponse>:

// Pre-processor: runs before the handler
public class AuditPreProcessor : IRequestPreProcessor<CreateOrderCommand>
{
    public async Task ProcessAsync(CreateOrderCommand request, CancellationToken cancellationToken)
    {
        _logger.LogInformation("Creating order for customer {CustomerId}", request.CustomerId);
        await _auditService.LogRequestAsync(request);
    }
}

// Post-processor: runs after the handler
public class OrderMetricsPostProcessor : IRequestPostProcessor<CreateOrderCommand, OrderResult>
{
    public async Task ProcessAsync(
        CreateOrderCommand request,
        OrderResult response,
        CancellationToken cancellationToken)
    {
        _metrics.RecordOrderCreated(response.Total);
        await _cacheService.InvalidateOrderCacheAsync(request.CustomerId);
    }
}

// Register processor behaviors
builder.Services
    .AddFSMediator()
    .AddRequestPreProcessorBehavior()     // Auto-discovers all IRequestPreProcessor<T> implementations
    .AddRequestPostProcessorBehavior();   // Auto-discovers all IRequestPostProcessor<T, TResponse> implementations

📋 Comprehensive Examples

Example 1: E-commerce Order Processing

// Commands and Queries
public record CreateOrderCommand(int CustomerId, List<OrderItem> Items) : IRequest<OrderResult>;
public record GetOrderHistoryQuery(int CustomerId, int PageSize) : IStreamRequest<Order>;
public record OrderCreatedNotification(Order Order) : INotification;

// Handlers with built-in resilience
public class CreateOrderHandler : IRequestHandler<CreateOrderCommand, OrderResult>
{
    public async Task<OrderResult> HandleAsync(CreateOrderCommand request, CancellationToken cancellationToken)
    {
        // Business logic with automatic retry, circuit breaking, and logging
        var order = await _orderService.CreateOrderAsync(request.CustomerId, request.Items);
        
        // Publish notification - multiple handlers can process this
        await _mediator.PublishAsync(new OrderCreatedNotification(order));
        
        return new OrderResult(order.Id, order.Total);
    }
}

// Streaming for large datasets
public class GetOrderHistoryHandler : IStreamRequestHandler<GetOrderHistoryQuery, Order>
{
    public async IAsyncEnumerable<Order> HandleAsync(
        GetOrderHistoryQuery request, 
        [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        // Process orders one by one - memory efficient for large histories
        await foreach (var order in _repository.GetOrderHistoryStreamAsync(request.CustomerId))
        {
            yield return order;
        }
    }
}

// Multiple notification handlers
public class OrderCreatedEmailHandler : INotificationHandler<OrderCreatedNotification>
{
    public async Task HandleAsync(OrderCreatedNotification notification, CancellationToken cancellationToken)
    {
        await _emailService.SendOrderConfirmationAsync(notification.Order);
    }
}

public class OrderCreatedInventoryHandler : INotificationHandler<OrderCreatedNotification>
{
    public async Task HandleAsync(OrderCreatedNotification notification, CancellationToken cancellationToken)
    {
        await _inventoryService.UpdateStockAsync(notification.Order.Items);
    }
}

Example 2: Real-time Data Processing

// Configuration for high-throughput scenario
builder.Services
    .AddFSMediator()
    .AddStreamingPlatinumPackage(                     // Premium protection package
        resourcePreset: ResourceManagementPreset.HighPerformance,
        backpressurePreset: BackpressurePreset.RealTime)
    .AddStreamingHealthCheckBehavior<CustomHealthReporter>();

// Processing large datasets with full monitoring
public record ProcessSensorDataQuery(DateTime FromTime, string SensorType) : IStreamRequest<ProcessedReading>;

public class ProcessSensorDataHandler : IStreamRequestHandler<ProcessSensorDataQuery, ProcessedReading>
{
    public async IAsyncEnumerable<ProcessedReading> HandleAsync(
        ProcessSensorDataQuery request, 
        [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        // Stream millions of sensor readings with automatic:
        // - Memory management
        // - Performance monitoring  
        // - Health checking
        // - Backpressure handling
        // - Error recovery
        
        await foreach (var reading in _sensorRepository.GetReadingsStreamAsync(request.FromTime, request.SensorType))
        {
            var processed = await _processor.ProcessReadingAsync(reading);
            yield return processed;
        }
    }
}

⚙️ Configuration Reference

Pipeline Behaviors

Pipeline behaviors provide cross-cutting concerns and execute in registration order:

builder.Services
    .AddFSMediator()
    
    // Logging - always first for complete visibility
    .AddLoggingBehavior()
    
    // Performance monitoring
    .AddPerformanceBehavior(warningThresholdMs: 1000)
    
    // Resilience patterns
    .AddRetryBehavior(options => {
        options.MaxRetryAttempts = 3;
        options.Strategy = RetryStrategy.ExponentialBackoffWithJitter;
        options.MaxTotalRetryTime = TimeSpan.FromSeconds(30);
    })
    
    .AddCircuitBreakerBehavior(options => {
        options.FailureThresholdPercentage = 50.0;
        options.DurationOfBreak = TimeSpan.FromSeconds(30);
        options.MinimumThroughput = 5;
    })
    
    // Resource management
    .AddResourceManagementBehavior(ResourceManagementPreset.Balanced);

Validation Configuration

Configure request validation for both standard and streaming pipelines:

builder.Services
    .AddFSMediator()

    // Standard request validation
    .AddValidationBehavior()                          // Auto-discovers IValidator<T> implementations

    // Streaming request validation
    .AddStreamingValidationBehavior();                // Auto-discovers validators for stream requests

Auto-scan automatically registers all IValidator<> implementations found in the provided assemblies.

Exception Handling Configuration

Configure centralized exception handling for consistent error management:

builder.Services
    .AddFSMediator()

    // Standard exception handling
    .AddExceptionHandlingBehavior(options =>
    {
        options.RethrowUnhandledExceptions = true;    // Rethrow when no matching handler is found
    })

    // Streaming exception handling
    .AddStreamingExceptionHandlingBehavior();

Auto-scan automatically registers all IExceptionHandler<,,> implementations found in the provided assemblies.

Timeout Configuration

Configure timeout behavior using presets or custom options:

builder.Services
    .AddFSMediator()

    // Use a preset
    .AddTimeoutBehavior(TimeoutPreset.Standard)

    // Or configure manually
    .AddTimeoutBehavior(options =>
    {
        options.DefaultTimeout = TimeSpan.FromSeconds(45);
    })

    // Streaming timeout
    .AddStreamingTimeoutBehavior();

Pre/Post Processor Configuration

Register request pre-processors and post-processors:

builder.Services
    .AddFSMediator()

    // Pre-processors run before the handler
    .AddRequestPreProcessorBehavior()                 // Auto-discovers IRequestPreProcessor<T>

    // Post-processors run after the handler
    .AddRequestPostProcessorBehavior();               // Auto-discovers IRequestPostProcessor<T, TResponse>

Auto-scan automatically registers all IRequestPreProcessor<> and IRequestPostProcessor<,> implementations found in the provided assemblies.

Streaming Configuration

Configure streaming behaviors for different scenarios:

// For data processing workloads
builder.Services
    .AddFSMediator()
    .AddStreamingLoggingBehavior(options => {
        options.LogProgressEveryNItems = 10000;      // Log every 10k items
        options.LogProgressEveryNSeconds = 60;       // Log every minute
    })
    .AddStreamingBackpressureBehavior(BackpressurePreset.Analytics)
    .AddStreamingResourceManagementBehavior(ResourceManagementPreset.MemoryConstrained);

// For real-time applications  
builder.Services
    .AddFSMediator()
    .AddStreamingBackpressureBehavior(BackpressurePreset.RealTime)
    .AddStreamingHealthCheckBehavior(HealthCheckPreset.HighPerformance);

Backpressure Strategies

Choose the right strategy for your use case:

// Buffer strategy (default) - queue items when consumer is slow
builder.Services.AddStreamingBackpressureBehavior(options => {
    options.Strategy = BackpressureStrategy.Buffer;
    options.MaxBufferSize = 10000;
});

// Drop strategy - discard items when overwhelmed
builder.Services.AddStreamingBackpressureBehavior(options => {
    options.Strategy = BackpressureStrategy.Drop;
    options.PreferNewerItems = true; // Keep latest data
});

// Throttle strategy - slow down producer to match consumer
builder.Services.AddStreamingBackpressureBehavior(options => {
    options.Strategy = BackpressureStrategy.Throttle;
    options.MaxThrottleDelayMs = 1000; // Max 1 second delay
});

// Sample strategy - process only subset of items under pressure
builder.Services.AddStreamingBackpressureBehavior(options => {
    options.Strategy = BackpressureStrategy.Sample;
    options.SampleRate = 2; // Process every 2nd item under pressure
});

// Block strategy - halt producer until consumer catches up
builder.Services.AddStreamingBackpressureBehavior(options => {
    options.Strategy = BackpressureStrategy.Block;
});

Request/Response Interceptors

Add cross-cutting concerns with surgical precision:

// Custom request interceptor
public class SecurityInterceptor<TRequest, TResponse> : IRequestInterceptor<TRequest, TResponse>
    where TRequest : IRequest<TResponse>
{
    public async Task<TRequest> InterceptRequestAsync(TRequest request, CancellationToken cancellationToken)
    {
        // Add security checks, audit logging, etc.
        await _securityService.ValidateRequestAsync(request);
        return request;
    }
}

// Register interceptors
builder.Services
    .AddFSMediator()
    .AddRequestInterceptor<GetSensitiveDataQuery, SensitiveData, SecurityInterceptor<GetSensitiveDataQuery, SensitiveData>>();

🔧 Advanced Features

Custom Health Monitoring

Implement custom health reporting for your monitoring infrastructure:

public class ApplicationInsightsHealthReporter : IStreamHealthReporter
{
    public async Task ReportHealthAsync(StreamHealthMetrics metrics, CancellationToken cancellationToken)
    {
        _telemetryClient.TrackMetric("StreamThroughput", metrics.CurrentThroughput);
        _telemetryClient.TrackMetric("StreamMemoryUsage", metrics.CurrentMemoryUsage);
    }

    public async Task ReportCriticalIssueAsync(StreamHealthMetrics metrics, HealthWarning warning, CancellationToken cancellationToken)
    {
        _telemetryClient.TrackException(new Exception($"Stream health critical: {warning.Message}"));
    }
}

// Register custom health reporter
builder.Services.AddStreamingHealthCheckBehavior<ApplicationInsightsHealthReporter>();

Custom Backpressure Strategies

Define custom logic for handling producer-consumer speed mismatches:

builder.Services.AddStreamingBackpressureBehavior(options => {
    options.Strategy = BackpressureStrategy.Buffer; // Use existing strategy
    options.CustomBackpressureHandler = context => {
        if (context.Metrics.CurrentBufferSize > 50000)
        {
            // Custom logic: maybe alert operations team
            _alertService.SendAlert("High backpressure detected");
            
            // Or dynamically scale resources
            _scalingService.ScaleOutAsync();
        }
    };
});

📊 Available Presets

BackpressurePreset Options

BackpressurePreset.NoDataLoss      // Prioritizes data completeness
BackpressurePreset.HighThroughput  // Maximizes performance
BackpressurePreset.MemoryConstrained // For limited memory environments
BackpressurePreset.RealTime        // For real-time applications
BackpressurePreset.Analytics       // For data processing/analytics
BackpressurePreset.Balanced        // General purpose

HealthCheckPreset Options

HealthCheckPreset.HighPerformance   // Real-time monitoring
HealthCheckPreset.DataProcessing    // Batch operations
HealthCheckPreset.LongRunning       // Overnight jobs
HealthCheckPreset.RealTime          // User-facing streams
HealthCheckPreset.Development       // Testing/debugging

ResourceManagementPreset Options

ResourceManagementPreset.MemoryConstrained  // Containers/embedded
ResourceManagementPreset.HighPerformance    // Performance critical
ResourceManagementPreset.Balanced           // Most applications
ResourceManagementPreset.Development        // Debugging scenarios

CircuitBreakerPreset Options

CircuitBreakerPreset.Sensitive     // Quick failure detection
CircuitBreakerPreset.Balanced      // General purpose
CircuitBreakerPreset.Resilient     // High failure tolerance
CircuitBreakerPreset.Database      // Database operations
CircuitBreakerPreset.ExternalApi   // External service calls

RetryPreset Options

RetryPreset.Conservative   // Quick failure, minimal retries
RetryPreset.Aggressive     // Persistent retries
RetryPreset.Database       // Database-specific handling
RetryPreset.HttpApi        // HTTP/network operations

TimeoutPreset Options

Preset	Timeout	Use Case
QuickResponse	5s	Cache lookups, simple CRUD
Standard	30s	Database queries, API calls
LongRunning	2min	Reports, file processing
Batch	10min	Data migrations, bulk ops
ExternalApi	15s	Third-party API calls

📊 Performance Characteristics

FS.Mediator is designed for high-performance scenarios:

Feature	Performance Impact	Memory Usage	Best For
Basic Mediator	< 1μs overhead	Minimal	All scenarios
Streaming	~2-5μs per item	O(1) constant	Large datasets
Retry Behavior	Varies by strategy	Minimal	Unreliable services
Circuit Breaker	< 1μs when closed	~1KB per type	External dependencies
Backpressure	Adaptive	Configurable	High-throughput streams

Benchmarks

Real-world performance metrics on a standard development machine:

// Basic request/response
BenchmarkDotNet results:
|     Method |      Mean |   Error |  StdDev |
|----------- |----------:|--------:|--------:|
| SimpleRequest | 1.2 μs | 0.02 μs | 0.02 μs |

// Streaming throughput  
Items processed: 1,000,000
Total time: 2.3 seconds
Throughput: 434,782 items/second
Memory usage: 45 MB (constant)

🚧 Troubleshooting

Common Issues

Issue: Handlers not found

HandlerNotFoundException: No handler found for type 'MyRequest'

Solution: Ensure your handlers are in assemblies passed to AddFSMediator():

builder.Services.AddFSMediator(typeof(MyHandler).Assembly);

Issue: Streaming performance problems

Stream throughput below expected levels

Solution: Configure backpressure and resource management:

builder.Services
    .AddFSMediator()
    .AddStreamingBackpressureBehavior(BackpressurePreset.HighThroughput)
    .AddStreamingResourceManagementBehavior(ResourceManagementPreset.HighPerformance);

Issue: Memory usage growing during streaming

OutOfMemoryException during large stream processing

Solution: Enable resource management with appropriate presets:

builder.Services
    .AddFSMediator()
    .AddStreamingResourceManagementBehavior(ResourceManagementPreset.MemoryConstrained);

Best Practices

1. Always use streaming for large datasets - Don't load millions of records into memory
2. Configure resilience patterns early - Add retry and circuit breaker behaviors from the start
3. Monitor your streams - Use health check behaviors in production
4. Choose appropriate presets - Start with presets, customize only when needed
5. Test backpressure scenarios - Simulate high load to validate your configuration

Debugging Tips

Enable detailed logging to understand behavior execution:

builder.Services
    .AddFSMediator()
    .AddLoggingBehavior()  // Logs all requests/responses
    .AddStreamingLoggingBehavior(options => {
        options.LogProgressEveryNItems = 1000;      // Progress updates
        options.LogDetailedMetrics = true;          // Performance data
    });

📚 Documentation

For more detailed documentation, see:

• Getting Started Guide
• Streaming Operations
• Resilience Patterns
• Configuration Reference
• Performance Tuning
• Examples Repository

🤝 Contributing

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

Development Setup

git clone https://github.com/furkansarikaya/FS.Mediator.git
cd FS.Mediator
dotnet restore
dotnet build
dotnet test

📦 Packages

Package	Description	NuGet
FS.Mediator	Core library with all features

🗺️ Roadmap

• Performance Improvements: Zero-allocation streaming paths
• Additional Patterns: Saga pattern support for complex workflows
• Observability: OpenTelemetry integration
• Cloud Native: Kubernetes health checks integration
• AI/ML: Streaming ML pipeline support

🌟 Star History

If you find this library useful, please consider giving it a star on GitHub! It helps others discover the project.

Made with ❤️ by Furkan Sarıkaya

---

Support

If you encounter any issues or have questions:

1. Check the troubleshooting section
2. Search existing GitHub issues
3. Create a new issue with detailed information
4. Join our community discussions

Happy coding! 🚀