Documentation

Messages and handlers

Commands, queries and events: who can have how many handlers, and why.

The three kinds

Message	Base	Handler	How many handlers
Command without a result	`Command`	`ICommandHandler<TCommand>`	exactly 1
Command with a result	`Command<TResult>`	`ICommandHandler<TCommand, TResult>`	exactly 1
Query	`Query<TResult>`	`IQueryHandler<TQuery, TResult>`	exactly 1
Streaming query	`StreamQuery<TItem>`	`IStreamQueryHandler<TQuery, TItem>`	exactly 1
Event	`Event`	`IEventHandler<TEvent>`	0 or more

The cardinality is not decoration. A command and a query have one owner: if two handlers compete for the same command, registration fails at start-up with DuplicateHandlerException, rather than one of them winning silently depending on the order the assembly was scanned in. An event is a notification: any number of interested parties, and none of them is not an error.

Metadata: IMessage

Every message carries three fields, inherited from IMessage. That is what lets the logging pipeline open a scope with the message id without knowing the concrete type, and the enrichment pipeline fill in authorship.

public interface IMessage
{
    Guid Id { get; }
    DateTimeOffset CreatedOn { get; }

    // Settable on purpose: the enrichment pipeline fills it from
    // ICurrentUserAccessor when it is empty.
    string? CreatedBy { get; set; }
}

A value that is already set is respected. A command replayed by a background job can declare the original author, and enrichment does not overwrite it.

Writing messages

Use init on the properties. A message is input data, not a mutable object — the one exception is CreatedBy, which the pipeline fills.

public sealed class CreateProduct : Command<Guid>
{
    public Guid CategoryId { get; init; }
    public string Sku { get; init; } = string.Empty;
    public string Name { get; init; } = string.Empty;
    public decimal Price { get; init; }
}

public sealed class DeleteProduct : Command
{
    public Guid ProductId { get; init; }
}

public sealed class GetProduct : Query<ProductDetails?>
{
    public Guid ProductId { get; init; }
}

public sealed class ProductRepriced(Guid productId, decimal from, decimal to) : Event
{
    public Guid ProductId { get; } = productId;
    public decimal From { get; } = from;
    public decimal To { get; } = to;
}

Per-message options

Three attributes change what the pipelines do with one specific message, without touching the global configuration. They are read once per type and cached.

Attribute	Effect
`[NoTransaction]`	The command is still saved at the end, as always, but without a transaction around it. For long imports that write in batches.
`[SkipValidation]`	The validation pipeline does not run for this message — an internal command, raised by code that already validated everything.
`[MessageTimeout(seconds)]`	The handler's token is cancelled at the deadline, and the message fails with `TimeoutException`. `AddNexo<TContext>` already includes the pipeline; assembling by hand, `AddTimeoutPipeline()`.

[MessageTimeout(5)]
public sealed class SearchProducts : PagedQuery<ProductSummary>
{
    public string? Term { get; init; }
}

// Assembling the pipelines by hand (AddNexo<TContext> already includes it):
services.AddNexo(nexo => nexo
    .AddLoggingPipeline()
    .AddTimeoutPipeline()
    .AddUnitOfWorkPipeline());

One handler, several messages

A class can implement as many handler interfaces as it likes, and is registered once per interface. That keeps together what changes together: every write for an aggregate in one file, with the dependencies declared once.

public sealed class ProductCommandHandlers(
    ICreate<Product> create,
    IDelete<Product> delete,
    IReadByExternalId<Product> byExternalId)
    : ICommandHandler<CreateProduct, Guid>,
      ICommandHandler<RenameProduct>,
      ICommandHandler<DeleteProduct>
{
    public async Task<Guid> HandleAsync(CreateProduct command, CancellationToken ct)
    {
        var product = await create.CreateAsync(new Product(command.Sku, command.Price), ct);
        return product.ExternalId;
    }

    public async Task HandleAsync(RenameProduct command, CancellationToken ct)
    {
        // A read by key comes back tracked: mutating it is the whole update.
        var product = await byExternalId.ReadRequiredAsync(command.ProductId, ct);
        product.Rename(command.Name);
    }

    public async Task HandleAsync(DeleteProduct command, CancellationToken ct)
    {
        var product = await byExternalId.ReadRequiredAsync(command.ProductId, ct);
        await delete.DeleteAsync(product, ct);
    }
}

Dispatching

public interface IMediator
{
    Task SendAsync(ICommand command, CancellationToken ct = default);
    Task<TResult> SendAsync<TResult>(ICommand<TResult> command, CancellationToken ct = default);
    Task<TResult> FetchAsync<TResult>(IQuery<TResult> query, CancellationToken ct = default);
    IAsyncEnumerable<TItem> StreamAsync<TItem>(IStreamQuery<TItem> query, CancellationToken ct = default);
    Task BroadcastAsync(IEvent @event, CancellationToken ct = default);
}

A streaming query hands back items one at a time, as the database produces them, instead of materialising the whole list — for exporting a large CSV, feeding a report, or answering with IAsyncEnumerable from an endpoint:

public sealed class ExportProducts : StreamQuery<ProductRow>;

internal sealed class ExportProductsHandler(IQueryable<Product> products)
    : IStreamQueryHandler<ExportProducts, ProductRow>
{
    public IAsyncEnumerable<ProductRow> HandleAsync(ExportProducts query, CancellationToken ct) =>
        products
            .OrderBy(p => p.Id)
            .Select(p => new ProductRow(p.ExternalId, p.Sku, p.Price))
            .AsAsyncEnumerable();
}

// At the endpoint: ASP.NET Core serialises the stream as it arrives.
app.MapGet("/products/export", (IMediator mediator, CancellationToken ct) =>
    mediator.StreamAsync(new ExportProducts(), ct));

Pipelines see streams too (OnStreamAsync): logging, validation and telemetry all apply. The unit of work does not — a stream is a read.

IMediator is scoped. Every request, every message consumed from a queue, every iteration of a worker should run in its own DI scope — the scope is what defines the boundary of the unit of work.

// In a worker, create the scope explicitly.
await using var scope = serviceProvider.CreateAsyncScope();
var mediator = scope.ServiceProvider.GetRequiredService<IMediator>();

await mediator.SendAsync(new ImportProducts { /* ... */ }, ct);

Registration and scanning

Scanning picks up concrete, non-generic classes that implement any handler interface — including internal types, which is where most handlers should be.

services.AddNexo(nexo => nexo
    // One assembly, by a marker type.
    .AddHandlersFromAssemblyOf<Program>()

    // Several at once.
    .AddHandlersFromAssemblies(typeof(Program).Assembly, typeof(Billing).Assembly)

    // With a filter: leaves out what you want to register by hand.
    .AddHandlersFromAssemblyOf<Program>(filter: t => !t.Namespace!.Contains("Legacy"))

    // One specific handler, from a third-party assembly.
    .AddHandler<VendorWebhookHandler>());

The default lifetime is Scoped, which is what matches the unit of work they write through. Adjust it with nexo.HandlerLifetime if you need to. An assembly holding a type that will not load — an optional dependency that is missing — does not bring the scan down: the types that load are registered, the rest are left out.

The Nexo package carries a source generator that does the scanning at compile time. It writes an AddGeneratedHandlers() into your assembly with every handler registered by name — no reflection at start-up — and prepares the dispatch for every message type, which would otherwise be built with MakeGenericType on first use. It is the route to trimming and native AOT:

services.AddNexo<CatalogDbContext>(nexo => nexo
    .AddGeneratedHandlers()     // generated for this assembly; instead of AddHandlersFromAssemblyOf<Program>()
    .AddValidatorsFromAssemblyOf<Program>());

Diagnostic	When	Severity
`NEXO001`	Two handlers for the same command or query — the error scanning would only raise at start-up shows up in the build.	error
`NEXO002`	A nested `private` handler, which the generated code cannot reach.	warning

Events: sequential or parallel

By default an event's handlers run one after another, in registration order. That is not conservatism: they share the scope, and therefore the DbContext, which is not thread-safe.

services.AddNexo(nexo => nexo
    .ConfigureMediator(mediator => mediator.EventDispatchMode = EventDispatchMode.Parallel));

How dispatch works underneath

SendAsync(ICommand) only knows the interface. To call the typed handler without reflection on the hot path, Nexo builds an *invoker* closed over the message type the first time it sees that type, and caches it forever.

From then on a dispatch is a ConcurrentDictionary lookup plus a virtual call. The pipeline chain is composed from ordinary delegates. There is no Expression.Compile, and no MakeGenericMethod per call.