AwadyLab.Mediator 1.7.2

AwadyLab.Mediator

A high-performance, low-allocation Mediator, Async Streaming, and Pub/Sub Messaging engine for .NET 10+.

Overview

AwadyLab.Mediator provides a clean, decoupled architecture for modern .NET applications. Dispatch is organized into three focused pillars unified by an ergonomic facade:

flowchart TD
    Caller(["Caller / API Endpoint"]) --> Facade{{"IMediator"}}
    
    Facade --> Req["IRequestMediator<br/>(Request / Response)"]
    Facade --> Str["IStreamMediator<br/>(Async Streaming)"]
    Facade --> Notif["INotificationPublisher<br/>(Pub / Sub)"]
    
    Req --> Pipe["Pipeline Behaviors<br/>&amp; Short-Circuit"]
    Str --> Stream["IAsyncEnumerable&lt;T&gt;<br/>&amp; Decorators"]
    Notif --> Deliv{"Delivery Mode"}
    
    Deliv --> Direct["Direct<br/>(In-Process)"]
    Deliv --> Queue["Queue<br/>(Background Channel)"]
    Deliv --> Broker["Broker<br/>(Distributed Pub/Sub)"]
    
    Pipe --> H1{{"IRequestHandler&lt;,&gt;"}}
    Stream --> H2{{"IStreamRequestHandler&lt;,&gt;"}}
    Direct --> H3{{"INotificationHandler&lt;&gt;"}}
    Queue --> H3
    Broker --> H4{{"IExternalNotificationHandler&lt;&gt;"}}

Key Highlights

• Zero-Allocation Short-Circuit: When no pipeline behaviors are configured, requests bypass middleware overhead and invoke handlers directly without delegate closures.
• Cached Reflection Invokers: Eliminates dynamic MethodInfo.Invoke on hot dispatch paths — each (handler, request/response) or (handler, notification) pair gets a strongly-typed delegate built once at startup from a closed generic method, not a compiled expression tree (which can't legally call an internal handler type's members).
• Automatic DI Scope Management: Automatically reuses HttpContext.RequestServices when running inside web requests, and provisions isolated IServiceScope instances in background jobs and parallel dispatches.
• Async Stream Processing: Built-in support for IAsyncEnumerable<TResponse> with scope persistence across stream lifetimes, early request vetoing (yield break), and per-item response filtering.
• Multi-Tier Notification Engine: Supports in-process direct publishing, in-memory durable channel queues with exponential backoff retries, and cross-service broker messaging (RabbitMQ, Redis) with security sanitization.

Installation

dotnet add package AwadyLab.Mediator

Quickstart

Register in DI (Program.cs)

using AwadyLab.Mediator;

var builder = WebApplication.CreateBuilder(args);

// Register mediator scanning candidate assemblies
builder.Services.AddMediator([typeof(Program).Assembly], options =>
{
    // Optional: Add global pipeline behaviors
    options.ExecutorPipelines.Add(typeof(LoggingPipeline<,>));
    options.ExecutorPipelines.Add(typeof(ValidationPipeline<,>));
});

var app = builder.Build();

1. Request / Response (IRequestMediator)

Single-request, single-response execution designed for standard CQRS commands and queries.

Defining Requests & Handlers

All response contracts are constrained to IResponse. IResponse serves as a marker interface that promotes clear architectural boundaries, guarantees consistency, and allows reflection and tooling to query and inspect all response types across your domain and assemblies. We strongly encourage adopting this pattern for all domain responses.

using AwadyLab.Mediator.Abstraction;

// 1. Define Request and Response records — a reference type implements IResponse directly.
public record GetUserQuery(int UserId) : IRequest<UserDto>;

public record UserDto(int Id, string Name, string Email) : IResponse;

// 2. Implement Handler
public class GetUserHandler(IUserRepository repo) : IRequestHandler<GetUserQuery, UserDto>
{
    public async Task<UserDto> Execute(GetUserQuery request, CancellationToken cancellationToken)
    {
        var user = await repo.FindByIdAsync(request.UserId, cancellationToken)
            ?? throw new KeyNotFoundException($"User {request.UserId} not found");

        return new UserDto(user.Id, user.Name, user.Email);
    }
}

Value Responses (ValueResponse<T>)

A response type you don't own the declaration of — a primitive, an enum, or a value type from another library (e.g., int, bool, Guid) — cannot be given : IResponse directly. ValueResponse<T> (T : struct) wraps it instead, providing implicit bidirectional conversions so you can return values naturally:

public record CountUsersQuery : IRequest<ValueResponse<int>>;

public class CountUsersHandler(IUserRepository repo) : IRequestHandler<CountUsersQuery, ValueResponse<int>>
{
    // int converts to ValueResponse<int> implicitly — no `new ValueResponse<int>(...)` needed.
    public async Task<ValueResponse<int>> Execute(CountUsersQuery request, CancellationToken cancellationToken) =>
        await repo.CountAsync(cancellationToken);
}

Void-like Requests (Unit)

For commands that return no payload, use Unit:

public record DeleteUserCommand(int UserId) : IRequest<Unit>;

public class DeleteUserHandler(IUserRepository repo) : IRequestHandler<DeleteUserCommand, Unit>
{
    public async Task<Unit> Execute(DeleteUserCommand request, CancellationToken cancellationToken)
    {
        await repo.DeleteAsync(request.UserId, cancellationToken);
        return Unit.Value;
    }
}

Pipeline Behaviors (IPipelineBehaviorHandler<,>)

Pipelines execute in the exact order that they are registered in options.ExecutorPipelines. Standard cross-cutting concerns like logging, execution measurement (timing), and validation are standard and trivial to chain:

options.ExecutorPipelines.Add(typeof(LoggingPipeline<,>));              // Runs 1st: logs entry and exit
options.ExecutorPipelines.Add(typeof(ExecutionMeasurementPipeline<,>)); // Runs 2nd: measures execution latency
options.ExecutorPipelines.Add(typeof(ValidationPipeline<,>));           // Runs 3rd: validates incoming command

When a request is executed, it passes through LoggingPipeline and ExecutionMeasurementPipeline before next() enters ValidationPipeline. On completion, the response unwinds in reverse order. Order your cross-cutting concerns deliberately (e.g., global exception handling outermost, logging second, timing third, validation fourth).

public class LoggingPipeline<TRequest, TResponse>(ILogger<LoggingPipeline<TRequest, TResponse>> logger)
    : IPipelineBehaviorHandler<TRequest, TResponse>
    where TRequest : IRequest<TResponse>
    where TResponse : IResponse
{
    public async Task<TResponse> HandleAsync(
        TRequest message,
        MessageHandlerDelegate<TResponse> next,
        CancellationToken cancellationToken)
    {
        logger.LogInformation("Handling {RequestName}", typeof(TRequest).Name);
        var response = await next();
        logger.LogInformation("Handled {RequestName}", typeof(TRequest).Name);
        return response;
    }
}

2. Asynchronous Streaming (IStreamMediator)

Streams large datasets, real-time events, or LLM tokens efficiently using IAsyncEnumerable<TResponse>.

Defining Stream Requests & Handlers

using System.Runtime.CompilerServices;
using AwadyLab.Mediator.Abstraction;

public record StreamMetricsRequest(int DeviceId) : IStreamRequest<MetricPoint>;

// A stream's TResponse implements IResponse too, same as a request/response call's — ValueResponse<T> is
// the escape hatch for a primitive or other struct value that can't implement IResponse itself.
public record MetricPoint(DateTime Timestamp, double Value) : IResponse;

public class StreamMetricsHandler : IStreamRequestHandler<StreamMetricsRequest, MetricPoint>
{
    public async IAsyncEnumerable<MetricPoint> ExecuteAsync(
        StreamMetricsRequest request,
        [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        for (var i = 0; i < 100; i++)\n        {
            cancellationToken.ThrowIfCancellationRequested();
            await Task.Delay(50, cancellationToken);
            yield return new(DateTime.UtcNow, Random.Shared.NextDouble());
        }
    }
}

Stream Request & Response Decorators

Configure global streaming interceptors via MediatorOptions using pattern matching:

builder.Services.AddMediator([typeof(Program).Assembly], options =>
{
    // Request Decorator: Vetoes or modifies request BEFORE resolving handlers or opening scopes
    options.StreamRequestDecorator = async (req, _) => req switch
    {
        StreamMetricsRequest { DeviceId: < 0 } => null, // Returning null triggers 'yield break' immediately
        _ => req
    };

    // Response Decorator: Filters or transforms individual emitted items
    options.StreamResponseDecorator = async (item, _) => item switch
    {
        MetricPoint { Value: < 0.1 } => null, // Returning null skips only this item (continue)
        _ => item
    };
});

• Scope Lifetime: The DI scope stays active for the entire duration of stream enumeration and disposes when the consumer finishes await foreach.

3. Pub/Sub Notifications (INotificationPublisher)

The notification engine distributes events to one or more subscribers using three delivery modes and configurable execution semantics:

Delivery Modes & Execution Types

• Delivery Modes:

  1. Direct (NotificationDelivery.Direct): In-process dispatch immediately executing local handlers within the current application.
  2. Queue (NotificationDelivery.Queue): In-memory background channel queue drained asynchronously by worker loops with exponential backoff retries and dead-letter sinks.
  3. Broker (NotificationDelivery.Broker): Distributed pub/sub across microservices (RabbitMQ, Redis) with automatic reconnect, serialization guards, and external topic routing.

• Execution Types (Direct Mode):

  • Parallel (ExecuteParallel = true, Default): All subscribed handlers execute concurrently using Task.WhenEach. Each handler is automatically provisioned with an isolated IServiceScope, eliminating multi-threaded race conditions on scoped dependencies like EF Core DbContext.
  • Sequential (ExecuteParallel = false): Handlers execute sequentially in registration order, sharing the Ambient Scope, and aborting immediately if cancellation is requested.

• What is an Ambient Scope?: An Ambient Scope refers to the resolution scope already active in the caller's context — most commonly HttpContext.RequestServices during an active ASP.NET Core web request. In sequential execution, handlers share this ambient scope so they can participate in the caller's active Unit of Work (such as sharing a single EF Core DbContext instance or per-request caches). When running outside of an active web request (such as in a background hosted service or queue worker), the mediator provisions a dedicated scope for the dispatch and cleans it up upon completion.

• Extensibility & Customization Interfaces: You can customize or override the default behavior of the notification engine by registering your own implementations of these core interfaces in DI:

  Interface	Default Implementation	Purpose
  IMessageQueue	ChannelMessageQueue	Backing queue abstraction for NotificationDelivery.Queue. Defaults to an internal bounded System.Threading.Channels.Channel.
  IDeadLetterSink	LoggingDeadLetterSink	Receives poison envelopes that exhaust all configured retry attempts. Override to persist messages to a database, cloud storage, or alert system.
  INotificationErrorHandler	LoggingNotificationErrorHandler	Hook invoked on each failed dispatch attempt during retry backoff. Override to record Prometheus/OpenTelemetry metrics or emit telemetry.
  IMessageSerializer	SystemTextJsonMessageSerializer	Serializes and deserializes payloads for external broker transports. Override to use MessagePack, Protobuf, etc.
  IMessageBroker	None (supplied by transport packages)	Broker transport contract for NotificationDelivery.Broker (implemented by AwadyLab.Mediator.RabbitMQ and AwadyLab.Mediator.Redis).

• Configurable Defaults:

  • The global fallback delivery mode is set by options.DefaultNotificationDelivery (defaults to Direct).
  • In-memory queue parameters (capacity, worker count, retries, jitter) are configured via options.UseNotificationQueue(...).
  • External broker subscription defaults are configured via options.UseExternalSubscriptionDefaults(...).
  • Any individual publish call can override options via the Action<PublishOptions>? configure parameter.

public record OrderPlacedNotification(Guid OrderId, decimal Amount) : INotification;

1. Direct Delivery (In-Process)

Dispatches immediately within the caller's process. You can select between two execution modes:

Parallel Mode (ExecuteParallel = true, Default)

flowchart LR
    Pub(["Publisher"]) --> H1["Handler A<br/>(Isolated Scope 1)"]
    Pub --> H2["Handler B<br/>(Isolated Scope 2)"]
    Pub --> H3["Handler C<br/>(Isolated Scope 3)"]

Sequential Mode (ExecuteParallel = false)

flowchart LR
    Pub(["Publisher"]) --> H1["Handler A"]
    H1 --> H2["Handler B"]
    H2 --> H3["Handler C"]
    
    subgraph AmbientScope ["Shared Ambient Scope"]
        H1
        H2
        H3
    end

await mediator.Publish(new OrderPlacedNotification(orderId, 150m), options =>
{
    options.Delivery = NotificationDelivery.Direct;
    options.ExecuteParallel = true; // Concurrent dispatch with isolated DI scope per handler
});

• Parallel Execution (Task.WhenEach): Handlers run concurrently with independent DI scopes, preventing concurrency conflicts on scoped dependencies like Entity Framework DbContext.
• Sequential Execution: Runs handlers sequentially in ambient scope, short-circuiting immediately if cancellation is requested.

2. Queue Delivery (Durable In-Memory Channel)

flowchart LR
    Pub(["Publisher"]) --> Q[("Channel Queue")]
    Q --> Pump["Worker Pump"]
    Pump --> Handlers["Handlers"]
    Handlers -. "Failure & Retries Exhausted" .-> DLQ[/"Dead-Letter Sink"/]

Offloads notifications to an asynchronous background channel with worker loops:

await mediator.Publish(new OrderPlacedNotification(orderId, 150m), options =>
{
    options.Delivery = NotificationDelivery.Queue;
});

• Default Backing Queue: The mediator registers an internal, bounded System.Threading.Channels.Channel via ChannelMessageQueue.
• Using a Custom Queue (IMessageQueue): To replace the in-memory channel with a persistent database or disk-backed queue, implement IMessageQueue and register it in DI:

public class CustomDatabaseMessageQueue : IMessageQueue
{
    public async ValueTask EnqueueAsync(NotificationEnvelope envelope, CancellationToken cancellationToken)
    {
        // Persist envelope to a database table or custom storage
    }

    public async IAsyncEnumerable<DequeuedMessage> DequeueAllAsync(
        [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        // Poll or stream envelopes from storage, yielding DequeuedMessage with an IMessageAckToken
        yield break;
    }
}

// In DI configuration:
builder.Services.AddSingleton<IMessageQueue, CustomDatabaseMessageQueue>();

• Retry Budget & Backoff: Configured via options.QueueOptions with exponential backoff and randomized jitter.
• Failure Handling: Failed attempts notify INotificationErrorHandler and route to IDeadLetterSink when retries are exhausted.

3. Broker Delivery (Cross-Service Pub/Sub)

flowchart LR
    Pub(["Publisher"]) --> Broker[("Message Broker<br/>RabbitMQ / Redis")]
    Broker --> Top["Topic / Stream"]
    Top --> Sub1["Billing Service Handler"]
    Top --> Sub2["Inventory Service Handler"]

Broadcasts notifications across distributed microservices. Requires the corresponding official extension transport package:

Package	Transport	Capabilities	Status
AwadyLab.Mediator.RabbitMQ	RabbitMQ (AMQP 0-9-1)	Durable queues, topic/fanout exchanges, publisher confirms, channel pooling
AwadyLab.Mediator.Redis	Redis Streams	Consumer groups, auto-claim for crashed workers, approximate stream trimming

Register the transport package in the AddMediator configure callback:

using AwadyLab.Mediator.RabbitMQ;
using AwadyLab.Mediator.RabbitMQ.Options;

builder.Services.AddMediator([typeof(Program).Assembly], options =>
{
    options.UseRabbitMq(new RabbitMqOptions
    {
        ConnectionString = "amqp://guest:guest@localhost:5672/",
        Broker = new RabbitMqBrokerOptions() // enables NotificationDelivery.Broker
    });
});

Customizing Broker Behavior & Core Interfaces

To customize broker messaging, the following interfaces are available:

• IMessageBroker: Core transport contract implemented by AwadyLab.Mediator.RabbitMQ and AwadyLab.Mediator.Redis. You can implement IMessageBroker (and IReconnectableMessageBroker) to support other brokers like Azure Service Bus, AWS SQS/SNS, or Apache Kafka.
• IExternalNotificationDefinition<T>: Declares where a notification is published (exchange or stream) as an external contract.
• IExternalNotificationHandler<T>: Implemented by consuming services, declaring their subscription QueueName.
• IMessageSerializer: Serializes wire payloads (default: SystemTextJsonMessageSerializer). Replace by registering your own IMessageSerializer in DI (e.g., MessagePack or Protobuf).
• IDeadLetterSink: Captures permanently failed envelopes after retries are exhausted.

using AwadyLab.Mediator.Abstraction.Messaging;
using AwadyLab.Mediator.Models.Options;

public sealed class OrderPlacedDefinition : IExternalNotificationDefinition<OrderPlacedNotification>
{
    public void Define(ExternalNotificationOptions options)
    {
        options.RabbitMq.Exchange = "orders.placed";
        options.Redis.Stream = "orders.placed";
    }
}

[ExternalSubscription(Prefetch = 20, MaxRetries = 5)]
public class BillingOrderHandler(IBillingService billing) : IExternalNotificationHandler<OrderPlacedNotification>
{
    public static string QueueName => "billing-service";

    public async Task Handle(OrderPlacedNotification notification, CancellationToken cancellationToken) =>
        await billing.ProcessInvoiceAsync(notification.OrderId, notification.Amount, cancellationToken);
}

• Security Sanitization: Message headers automatically strip CRLF injection characters and Unicode directional override attacks.
• Type Security: Deserialization is constrained to registered types to prevent arbitrary type injection.
• Auto-Resubscribe: External subscriptions automatically restore on network reconnection.

4. Notification Pipeline Behaviors (INotificationPipelineBehavior<>)

Standard cross-cutting concerns like logging, execution measurement (timing), and validation are standard and trivial in modern mediator libraries. However, pipeline behaviors in AwadyLab.Mediator excel when addressing complex distributed architecture problems — such as enforcing idempotency and message deduplication across asynchronous dispatches.

Because queue and broker deliveries operate under at-least-once delivery guarantees, redeliveries or rapid duplicate publishes can trigger redundant or hazardous side-effects (e.g., duplicate stock reorders or repeated charges). Rather than polluting every single notification handler with deduplication boilerplate, a custom INotificationPipelineBehavior<TNotification> cleanly intercepts dispatches, extracts message identity keys, checks a deduplication store, and short-circuits execution before handlers are invoked.

1. Define the Idempotent Attribute

An attribute placed on notification types to declaratively designate idempotency and specify which property acts as the unique partition key:

namespace SampleApp.ServiceA.Pipelines;

/// <summary>
/// Marks a notification type as requiring idempotent processing.
/// Duplicate publications with the same key will be deduplicated and skipped by the pipeline.
/// </summary>
[AttributeUsage(AttributeTargets.Class)]
public sealed class IdempotentAttribute(string? keyProperty = null) : Attribute
{
    public string? KeyProperty { get; } = keyProperty;
}

2. Implement the Generic Pipeline Behavior

The behavior implements INotificationPipelineBehavior<TNotification>. If marked with [Idempotent], it extracts the key (by configured property name or standard conventions: Id, Key, Sku), checks the deduplication store, and short-circuits execution without calling await next():

using System.Collections.Concurrent;
using System.Reflection;
using AwadyLab.Mediator.Abstraction;

namespace SampleApp.ServiceA.Pipelines;

/// <summary>
/// A notification pipeline behavior that enforces idempotency for notifications marked with
/// <see cref="IdempotentAttribute"/>.
/// </summary>
public sealed class IdempotentNotificationPipelineBehavior<TNotification>(
    ILogger<IdempotentNotificationPipelineBehavior<TNotification>> logger)
    : INotificationPipelineBehavior<TNotification>
    where TNotification : INotification
{
    // Thread-safe in-memory store tracking processed notification keys.
    // In distributed multi-replica environments, this can be backed by Redis or an external distributed cache.
    private static readonly ConcurrentDictionary<string, byte> ProcessedKeys = new();

    public async Task HandleAsync(
        TNotification notification,
        NotificationHandlerDelegate next,
        CancellationToken cancellationToken)
    {
        var attribute = typeof(TNotification).GetCustomAttribute<IdempotentAttribute>();
        if (attribute is null)
        {
            // Unmarked notification: pass straight through without idempotency checks
            await next();
            return;
        }

        var key = ExtractKey(notification, attribute.KeyProperty);
        var cacheKey = $"{typeof(TNotification).Name}:{key}";

        if (!ProcessedKeys.TryAdd(cacheKey, 0))
        {
            logger.LogWarning(
                "[Pipeline:Idempotent] Duplicate notification detected for {NotificationType} with key '{Key}'. Skipping handlers.",
                typeof(TNotification).Name, key);
            return; // Deduplicated: short-circuit without calling next()!
        }

        logger.LogInformation(
            "[Pipeline:Idempotent] First time seeing notification {NotificationType} with key '{Key}'. Dispatching to handlers.",
            typeof(TNotification).Name, key);

        await next();
    }

    private static string ExtractKey(TNotification notification, string? configuredProperty)
    {
        var type = typeof(TNotification);

        if (!string.IsNullOrWhiteSpace(configuredProperty))
        {
            var prop = type.GetProperty(configuredProperty, BindingFlags.Public | BindingFlags.Instance);
            if (prop != null)
                return prop.GetValue(notification)?.ToString() ?? string.Empty;
        }

        foreach (var candidate in new[] { "Id", "Key", "Sku" })
        {
            var prop = type.GetProperty(candidate, BindingFlags.Public | BindingFlags.Instance | BindingFlags.IgnoreCase);
            if (prop != null)
                return prop.GetValue(notification)?.ToString() ?? string.Empty;
        }

        return notification.ToString() ?? type.Name;
    }
}

3. Mark Notifications Declaratively

// Deduplicate stock reorder notifications using the 'Sku' property
[Idempotent(nameof(Sku))]
public sealed record StockReorderRequestedNotification(string Sku, int Quantity) : INotification;

4. Register in DI via NotificationPipelines

Register the open-generic pipeline behavior in MediatorOptions:

builder.Services.AddMediator([typeof(Program).Assembly], options =>
{
    // Notification pipeline: runs around notification handlers across Direct and Queue dispatches
    options.NotificationPipelines.Add(typeof(IdempotentNotificationPipelineBehavior<>));
});

When duplicate notifications arrive (via Direct in-process dispatches or background channel queue pumps), the pipeline intercepts them and logs a warning while skipping handler execution entirely:

info: SampleApp.ServiceA.Pipelines.IdempotentNotificationPipelineBehavior[0]
      [Pipeline:Idempotent] First time seeing notification StockReorderRequestedNotification with key 'WIDGET-1'. Dispatching to handlers.
info: SampleApp.ServiceA.Handlers.StockReorderHandler[0]
      [Queue] Stock reorder processed off the background pump for WIDGET-1: 100 requested.

warn: SampleApp.ServiceA.Pipelines.IdempotentNotificationPipelineBehavior[0]
      [Pipeline:Idempotent] Duplicate notification detected for StockReorderRequestedNotification with key 'WIDGET-1'. Skipping handlers.

4. Unified Facade (IMediator)

Inject IMediator into API endpoints or Controllers. Each method simply forwards to the corresponding mediator (IRequestMediator, IStreamMediator, INotificationPublisher) — if you never register a handler for one of those surfaces, AddMediator still registers IMediator itself, and calling that method throws an InvalidOperationException naming the missing service instead of a NullReferenceException.

// CreateOrderCommand : IRequest<CreateOrderResult>, and CreateOrderResult : IResponse — defined the same
// way as GetUserQuery/UserDto above.
[ApiController]
[Route("api/[controller]")]
public class OrdersController(IMediator mediator) : ControllerBase
{
    [HttpPost]
    public async Task<IActionResult> CreateOrder([FromBody] CreateOrderCommand command, CancellationToken ct)
    {
        // 1. Request / Response
        var result = await mediator.Execute(command, ct);

        // 2. Publish Notification — the second positional parameter is Action<PublishOptions>?, so a
        // CancellationToken has to be passed by name.
        await mediator.Publish(new OrderPlacedNotification(result.OrderId, result.Total), cancellationToken: ct);

        return Ok(result);
    }

    [HttpGet("stream")]
    public IAsyncEnumerable<MetricPoint> StreamMetrics(CancellationToken ct) =>
        // 3. Async Streaming
        mediator.ExecuteAsync(new StreamMetricsRequest(42), ct);
}

5. Observability & Performance Metrics

AwadyLab.Mediator provides first-class extension points for deep performance monitoring and production observability. By combining Pipeline Behaviors, Error Handlers, and Dead-Letter Sinks, you can capture runtime metrics across direct in-process, background channel, and distributed broker dispatches:

• Success & Failure Counts: Track successful executions vs. thrown exceptions.
• Latency & Execution Duration: Measure average, minimum, and maximum handler execution times with high-precision Stopwatch.GetTimestamp().
• Retries & Backoff Events: Intercept transient delivery retries (willRetry == true) before messages are abandoned.
• Dead-Letter & Poison Messages: Capture messages that exhausted all retry attempts (IDeadLetterSink).
• Active In-Flight Workloads: Track concurrent active dispatches in real-time.
• OpenTelemetry & System.Diagnostics.Metrics: Export counters and histograms to Prometheus, Grafana, Datadog, or Azure Application Insights.

flowchart LR
    Disp["Dispatcher / Queue / Broker"] --> Pipe["PerformanceMetrics Pipeline<br/><i>measures duration &amp; in-flight</i>"]
    Pipe --> H["Handler(s)"]
    
    H -- "Success" --> M1["Record Duration &amp; Success"]
    H -- "Failure" --> Err["INotificationErrorHandler<br/><i>tracks retries &amp; backoff</i>"]
    Err -- "Retries Exhausted" --> DLQ["IDeadLetterSink<br/><i>tracks dead-letter count</i>"]

1. Performance Metrics Pipeline Behavior

Measures execution duration, tracks active in-flight count, and records success or failure:

public sealed class PerformanceMetricsNotificationPipelineBehavior<TNotification>(
    ServiceBMetrics metrics)
    : INotificationPipelineBehavior<TNotification>
    where TNotification : INotification
{
    public async Task HandleAsync(
        TNotification notification,
        NotificationHandlerDelegate next,
        CancellationToken cancellationToken)
    {
        var typeName = typeof(TNotification).Name;
        metrics.RecordExecutionStart(typeName);
        var start = Stopwatch.GetTimestamp();

        try
        {
            await next();
            var durationMs = Stopwatch.GetElapsedTime(start).TotalMilliseconds;
            metrics.RecordExecutionCompleted(typeName, durationMs, success: true);
        }
        catch (Exception ex)
        {
            var durationMs = Stopwatch.GetElapsedTime(start).TotalMilliseconds;
            metrics.RecordExecutionCompleted(typeName, durationMs, success: false);
            throw; // Re-throw to allow queue/broker retry mechanisms to engage
        }
    }
}

2. Tracking Retries via INotificationErrorHandler

Whenever a handler throws an exception during queue or broker dispatch, INotificationErrorHandler is notified with willRetry:

public sealed class MetricsNotificationErrorHandler(
    ServiceBMetrics metrics,
    ILogger<MetricsNotificationErrorHandler> logger)
    : INotificationErrorHandler
{
    public Task OnDispatchFailedAsync(
        NotificationEnvelope envelope,
        Exception exception,
        bool willRetry,
        CancellationToken cancellationToken)
    {
        metrics.RecordRetry(envelope.NotificationType.Name, envelope.Attempt, willRetry);
        return Task.CompletedTask;
    }
}

3. Tracking Poison Messages via IDeadLetterSink

When all configured retries are exhausted, the envelope is sent to IDeadLetterSink:

public sealed class MetricsDeadLetterSink(
    ServiceBMetrics metrics,
    ILogger<MetricsDeadLetterSink> logger)
    : IDeadLetterSink
{
    public Task SendAsync(
        NotificationEnvelope envelope,
        Exception lastException,
        CancellationToken cancellationToken)
    {
        metrics.RecordDeadLetter(envelope.NotificationType.Name);
        return Task.CompletedTask;
    }
}

4. Registering in DI

Register your custom metrics services and pipeline behaviors in Program.cs:

builder.Services.AddSingleton<ServiceBMetrics>();
builder.Services.AddSingleton<INotificationErrorHandler, MetricsNotificationErrorHandler>();
builder.Services.AddSingleton<IDeadLetterSink, MetricsDeadLetterSink>();

builder.Services.AddMediator([typeof(Program).Assembly], options =>
{
    // Register metrics pipeline behavior
    options.NotificationPipelines.Add(typeof(PerformanceMetricsNotificationPipelineBehavior<>));
});

Configuration Reference

Options Reference Table

The following tables list all configurable options across the mediator:

MediatorOptions

NotificationQueueOptions (Configured via UseNotificationQueue)

PublishOptions (Configured per Publish call)

Configuration Example

builder.Services.AddMediator([typeof(Program).Assembly], options =>
{
    // Custom predicate for discovered handler types
    options.TypeFilter = type => !type.Name.StartsWith("Test");

    // Pipeline behaviors for IRequestMediator
    options.ExecutorPipelines.Add(typeof(LoggingPipeline<,>));

    // Pipeline behaviors for INotificationPublisher
    options.NotificationPipelines.Add(typeof(NotificationAuditPipeline<>));

    // Default notification delivery mode (Direct, Queue, or Broker)
    options.DefaultNotificationDelivery = NotificationDelivery.Direct;

    // Queue & Retry Configuration
    options.UseNotificationQueue(queue =>
    {
        queue.Capacity = 5_000;
        queue.MaxConcurrency = 4;
        queue.MaxRetries = 3;
        queue.RetryDelay = TimeSpan.FromMilliseconds(200);
        queue.MaxRetryDelay = TimeSpan.FromSeconds(5);
    });

    // Streaming Decorators with pattern matching
    options.StreamRequestDecorator = async (req, _) => req;
    options.StreamResponseDecorator = async (resp, _) => resp;
});

License

Package by Mohamed Elawady

Copyright (c) Mohamed Elawady. All rights reserved.

This software is closed source and proprietary. Subject to the terms below,
you are granted a free, worldwide, non-exclusive license to:

  - Use this software, in source or compiled (NuGet package) form, in your
    own applications, including commercial applications, at no charge.

You may NOT:

  - Redistribute, sublicense, sell, or publish this software (or any
    modified version of it) as a standalone product, library, or package,
    including republishing it under a different name on NuGet.org or any
    other package repository.
  - Reverse engineer, decompile, or disassemble the compiled package except
    to the extent applicable law expressly permits this despite this
    limitation.
  - Remove or alter any copyright, trademark, or other proprietary notices.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.