MessageValidation 1.0.0

MessageValidation

A protocol-agnostic message validation pipeline for .NET — validate incoming messages from MQTT, RabbitMQ, Kafka, Azure Service Bus, NATS, or any messaging transport with DI integration, pluggable validation, and configurable failure handling.

Why?

Every message consumer faces the same challenge: raw bytes arrive, you deserialize them, and then you need to validate before processing. This logic is duplicated across every protocol and every project.

MessageValidation extracts this into a single, reusable pipeline:

Raw bytes → Deserialize → Validate → Handle (or dead-letter / log / skip)

The core has zero opinion on which messaging library or validation framework you use. Bring your own transport, bring your own validator.

Installation

dotnet add package MessageValidation

Quick Start

1. Define your message

public class TemperatureReading
{
    public string SensorId { get; set; } = "";
    public double Value { get; set; }
    public DateTime Timestamp { get; set; }
}

2. Implement a validator

public class TemperatureReadingValidator : IMessageValidator<TemperatureReading>
{
    public Task<MessageValidationResult> ValidateAsync(
        TemperatureReading message, CancellationToken ct = default)
    {
        var errors = new List<MessageValidationError>();

        if (string.IsNullOrWhiteSpace(message.SensorId))
            errors.Add(new("SensorId", "SensorId is required."));

        if (message.Value is < -50 or > 150)
            errors.Add(new("Value", "Value must be between -50 and 150."));

        return Task.FromResult(errors.Count == 0
            ? MessageValidationResult.Success()
            : MessageValidationResult.Failure(errors));
    }
}

3. Implement a handler

public class TemperatureHandler : IMessageHandler<TemperatureReading>
{
    public Task HandleAsync(
        TemperatureReading message, MessageContext context, CancellationToken ct = default)
    {
        // Only reached if validation passed
        Console.WriteLine($"[{context.Source}] Sensor {message.SensorId}: {message.Value}°C");
        return Task.CompletedTask;
    }
}

4. Implement a deserializer

using System.Text.Json;

public class JsonMessageDeserializer : IMessageDeserializer
{
    public object Deserialize(byte[] payload, Type targetType) =>
        JsonSerializer.Deserialize(payload, targetType)
        ?? throw new InvalidOperationException($"Failed to deserialize to {targetType.Name}");
}

5. Register services

builder.Services.AddMessageValidation(options =>
{
    options.MapSource<TemperatureReading>("sensors/+/temperature");
    options.DefaultFailureBehavior = FailureBehavior.Log;
});

builder.Services.AddMessageDeserializer<JsonMessageDeserializer>();
builder.Services.AddScoped<IMessageValidator<TemperatureReading>, TemperatureReadingValidator>();
builder.Services.AddMessageHandler<TemperatureReading, TemperatureHandler>();

6. Process messages

Feed messages into the pipeline from any transport:

var pipeline = serviceProvider.GetRequiredService<IMessageValidationPipeline>();

var context = new MessageContext
{
    Source = "sensors/living-room/temperature",
    RawPayload = payloadBytes
};

await pipeline.ProcessAsync(context);

Core Concepts

Source Mapping

Map source patterns (topics, queues, routing keys) to message types. Supports MQTT-style wildcards:

options.MapSource<TemperatureReading>("sensors/+/temperature");
options.MapSource<DeviceHeartbeat>("devices/#");

Failure Behaviors

Configure how validation failures are handled:

options.DefaultFailureBehavior = FailureBehavior.Custom;

// Register your custom handler
builder.Services.AddValidationFailureHandler<MyFailureHandler>();

Dead-Letter Queue

When FailureBehavior.DeadLetter is configured, the pipeline computes a dead-letter destination from DeadLetterPrefix + Source and delegates to an IDeadLetterHandler:

options.DefaultFailureBehavior = FailureBehavior.DeadLetter;
options.DeadLetterPrefix = "$dead-letter/"; // default — customize as needed

// Register your dead-letter handler
builder.Services.AddDeadLetterHandler<MyDeadLetterHandler>();

Implement the handler to publish the failed message to your transport's dead-letter destination:

public class MyDeadLetterHandler : IDeadLetterHandler
{
    public Task HandleAsync(DeadLetterContext context, CancellationToken ct = default)
    {
        // context.Destination       → "$dead-letter/sensors/room1/temperature"
        // context.OriginalContext    → the original MessageContext (source, payload, metadata)
        // context.ValidationResult  → the validation errors
        // context.Timestamp         → UTC time of the dead-letter decision
        Console.WriteLine($"Dead-lettering to {context.Destination}");
        return Task.CompletedTask;
    }
}

Resolution priority: When DeadLetter is active, the pipeline resolves handlers in this order:

1. IDeadLetterHandler — preferred (receives full DeadLetterContext with computed destination)
2. IValidationFailureHandler — backward-compatible fallback
3. Log warning — graceful degradation if no handler is registered

Abstractions

Architecture

The core library is transport-agnostic and validation-framework-agnostic. It defines contracts and a pipeline — adapters bring the implementations.

MessageValidation-Project/
├── MessageValidation/                          ← Core pipeline (zero opinion)
│   ├── Abstractions/
│   │   ├── IMessageValidationPipeline.cs          IMessageValidationPipeline
│   │   ├── IMessageValidator.cs                  IMessageValidator<T>
│   │   ├── IMessageHandler.cs                    IMessageHandler<T>
│   │   ├── IMessageDeserializer.cs               IMessageDeserializer
│   │   ├── IValidationFailureHandler.cs          IValidationFailureHandler
│   │   └── IDeadLetterHandler.cs                 IDeadLetterHandler
│   ├── Configuration/
│   │   ├── FailureBehavior.cs                    Log | DeadLetter | Skip | Throw | Custom
│   │   └── MessageValidationOptions.cs           Source-to-type mapping + wildcards
│   ├── Diagnostics/
│   │   └── MessageValidationMetrics.cs           System.Diagnostics.Metrics counters
│   ├── Models/
│   │   ├── MessageContext.cs                     Protocol-agnostic envelope
│   │   ├── MessageValidationResult.cs            Validation outcome
│   │   ├── MessageValidationError.cs             Single error record
│   │   └── DeadLetterContext.cs                  Dead-letter envelope (destination, errors, timestamp)
│   ├── Pipeline/
│   │   ├── MessageValidationPipeline.cs          Deserialize → Validate → Dispatch
│   │   └── MessageValidationException.cs         Thrown on FailureBehavior.ThrowException
│   └── DependencyInjection/
│       └── ServiceCollectionExtensions.cs        AddMessageValidation(), AddMessageHandler<,>(), AddDeadLetterHandler<>()
│
├── MessageValidation.DataAnnotations/          ← Validation adapter (DataAnnotations)
│   ├── DataAnnotationsMessageValidator.cs        Bridges DataAnnotations → IMessageValidator<T>
│   └── DependencyInjection/
│       └── ServiceCollectionExtensions.cs        AddMessageDataAnnotationsValidation()
│
├── MessageValidation.FluentValidation/         ← Validation adapter (FluentValidation)
│   ├── FluentValidationMessageValidator.cs       Bridges IValidator<T> → IMessageValidator<T>
│   └── DependencyInjection/
│       └── ServiceCollectionExtensions.cs        AddMessageFluentValidation()
│
├── MessageValidation.MqttNet/                  ← Transport adapter
│   ├── MqttClientExtensions.cs                   IMqttClient.UseMessageValidation()
│   ├── MqttServerExtensions.cs                   MqttServer.UseMessageValidation()
│   └── DependencyInjection/
│       └── ServiceCollectionExtensions.cs        AddMqttNetMessageValidation()
│
├── examples/
│   └── MessageValidation.Example/              ← Runnable console demo
│
└── README.md

Pipeline flow

flowchart LR
    Transport["🔌 Transport <br/> (MQTTnet, RabbitMQ,<br/> Kafka…)"]
    Deserializer["📦 IMessageDeserializer <br/> bytes → object"]
    Validator["✅ IMessageValidator&lt;T&gt; <br/> (FluentValidation,<br/> DataAnnotations…)"]
    Handler["⚡ IMessageHandler&lt;T&gt; <br/> (only if valid)"]
    Failure["⚠️ Failure Handler <br/> Log / Skip / Throw / Custom"]
    DeadLetter["💀 IDeadLetterHandler <br/> DeadLetterContext → <br/> transport DLQ"]

    Transport -->|raw bytes| Deserializer
    Deserializer -->|typed message| Validator
    Validator -->|valid| Handler
    Validator -->|invalid| Failure
    Validator -->|invalid + DeadLetter| DeadLetter

Adapter Packages

Roadmap

• v0.1 — Core pipeline, abstractions, DI integration, wildcard matching, FluentValidation adapter, MQTTnet transport adapter
• v0.2 — DataAnnotations adapter, System.Diagnostics.Metrics observability
• v0.3 — Dead-letter queue support (IDeadLetterHandler, DeadLetterContext, dead-letter metrics, backward-compatible fallback)
• v1.0 — RabbitMQ & Kafka adapters, source generators for AOT
• v2.0 — Middleware-style pipeline (Use, Map), Azure Service Bus adapter

Requirements

• .NET 10+

Author

• Romain OD
• @romainod | GitHub
• 🌐 www.devskillsunlock.com

License

MIT