MessageValidation: A Protocol-Agnostic Message Validation Pipeline for .NET

Every message consumer faces the same challenge: raw bytes arrive, you deserialize them, and then you have to validate before processing. This logic gets copy-pasted across every protocol and every project.

MessageValidation extracts that repeated work into a single, reusable pipeline — and stays completely agnostic about which transport or validation framework you use.

📦 GitHub: Romain-OD/MessageValidation · MIT · .NET 10

The Problem

Imagine you have three consumers — one on MQTT, one on RabbitMQ, one on Kafka. Each one ends up doing:

// ❌ Repeated in every consumer, every project
var message = JsonSerializer.Deserialize<OrderPlaced>(payload);
if (message.OrderId == Guid.Empty) { /* drop or log */ return; }
if (message.Amount <= 0)           { /* dead-letter  */ return; }
await _handler.HandleAsync(message);

The transport changes. The shape changes. The boilerplate does not.

The Solution: One Pipeline, Any Transport

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

Install the core package:

dotnet add package MessageValidation

The core has zero opinion on which messaging library or validation framework you use. Adapters bring the implementations.

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 when validation passes
        Console.WriteLine($"[{context.Source}] Sensor {message.SensorId}: {message.Value}°C");
        return Task.CompletedTask;
    }
}

4 — Register & run

builder.Services.AddMessageValidation(options =>
{
    options.MapSource<TemperatureReading>("sensors/+/temperature");
    options.DefaultFailureBehavior = FailureBehavior.DeadLetter;
});
builder.Services.AddMessageDeserializer<JsonMessageDeserializer>();
builder.Services.AddScoped<IMessageValidator<TemperatureReading>, TemperatureReadingValidator>();
builder.Services.AddMessageHandler<TemperatureReading, TemperatureHandler>();

// Feed any raw bytes into the pipeline
var pipeline = serviceProvider.GetRequiredService<IMessageValidationPipeline>();
await pipeline.ProcessAsync(new MessageContext
{
    Source     = "sensors/living-room/temperature",
    RawPayload = payloadBytes
});

Source Mapping & Wildcards

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

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

Five Failure Behaviors

When validation fails you choose what happens — no surprises, no hidden defaults:

options.DefaultFailureBehavior = FailureBehavior.DeadLetter;
options.DeadLetterPrefix = "$dead-letter/"; // default, override as needed
builder.Services.AddDeadLetterHandler<MyDeadLetterHandler>();

The DeadLetterContext carries the computed destination, the original MessageContext, the validation errors, and a UTC timestamp — everything you need to replay or audit the message.

Transport Adapters

The same AddMessageValidation(...) registration works across all transports. Only the transport line changes; handlers, validators, and dead-letter logic stay untouched.

MQTTnet

dotnet add package MessageValidation.MqttNet

builder.Services.AddMqttNetMessageValidation();

// Or hook directly onto an IMqttClient
mqttClient.UseMessageValidation(pipeline);

RabbitMQ

dotnet add package MessageValidation.RabbitMQ  // wraps RabbitMQ.Client 7.2.1

builder.Services.AddRabbitMqMessageValidation();

Kafka

dotnet add package MessageValidation.Kafka  // wraps Confluent.Kafka 2.13.2

builder.Services.AddKafkaMessageValidation();

Validation Adapters

Don't want to implement IMessageValidator<T> by hand? Drop in a validation adapter.

DataAnnotations

dotnet add package MessageValidation.DataAnnotations

public class OrderPlaced
{
    [Required] public Guid OrderId { get; set; }
    [Range(0.01, double.MaxValue)] public decimal Amount { get; set; }
}

// Bridges [Required], [Range], [MaxLength], … → IMessageValidator<T> automatically
builder.Services.AddMessageDataAnnotationsValidation();

FluentValidation

dotnet add package MessageValidation.FluentValidation

public class OrderPlacedValidator : AbstractValidator<OrderPlaced>
{
    public OrderPlacedValidator()
    {
        RuleFor(x => x.OrderId).NotEmpty();
        RuleFor(x => x.Amount).GreaterThan(0);
    }
}

// Bridges IValidator<T> → IMessageValidator<T> automatically
builder.Services.AddMessageFluentValidation();
builder.Services.AddScoped<IValidator<OrderPlaced>, OrderPlacedValidator>();

Architecture at a Glance

MessageValidation/                      ← Core pipeline (zero opinion)
│   Abstractions/
│   │   IMessageValidationPipeline      ← mockable contract
│   │   IMessageValidator<T>
│   │   IMessageHandler<T>
│   │   IMessageDeserializer
│   │   IDeadLetterHandler
│   Configuration/
│   │   FailureBehavior                 ← Log | DeadLetter | Skip | Throw | Custom
│   │   MessageValidationOptions        ← source-to-type mapping + wildcards
│   Diagnostics/
│       MessageValidationMetrics        ← System.Diagnostics.Metrics counters
│
MessageValidation.DataAnnotations/      ← validation adapter
MessageValidation.FluentValidation/     ← validation adapter
MessageValidation.MqttNet/              ← transport adapter
MessageValidation.RabbitMQ/             ← transport adapter
MessageValidation.Kafka/                ← transport adapter

The core defines contracts and the pipeline. Adapters bring the implementations. Nothing is coupled.

Observability Built In

The library ships System.Diagnostics.Metrics counters out of the box — plug them into your existing OpenTelemetry or Prometheus setup with no extra code.

Package Summary

All packages target .NET 10, are MIT licensed, and available on NuGet.org.

Roadmap

Getting Started

dotnet add package MessageValidation

🔗 GitHub — Romain-OD/MessageValidation
🔗 NuGet — MessageValidation
🔗 NuGet — All packages

Related in this series

• NATS Message Broker in .NET - subjects, queue groups and the NatsNet package in context
• Idempotency in .NET: the inbox pattern - validation rejects bad messages, idempotency handles repeated ones
• The Outbox Pattern in .NET - reliable publishing on the producer side
• Change Data Capture in PostgreSQL with .NET and Aspire - events sourced from the database log