Mediator

The mediator dispatches typed requests to their registered handlers in-process. It uses the same middleware pipeline model as the rest of Emit, so tracing, metrics, observers, and custom middleware all work without extra wiring.

Registration

Program.cs

builder.Services.AddEmit(emit =>
{
    emit.AddMediator(mediator =>
    {
        mediator.AddHandler<PlacePizzaOrderHandler>();
        mediator.AddHandler<GetPizzaStatusHandler>();
    });
});

AddMediator is an extension on EmitBuilder. You can call it alongside AddKafka or on its own; the two are independent.

Requests

A request is a plain class or record that implements IRequest (fire-and-forget) or IRequest<TResponse> (returns a value):

public record PlacePizzaOrder(
    string CustomerId,
    string[] Toppings,
    string Address) : IRequest;

public record GetPizzaStatus(Guid PizzaId) : IRequest<PizzaStatusDto>;

Handlers

Implement IRequestHandler<TRequest> or IRequestHandler<TRequest, TResponse>:

public class PlacePizzaOrderHandler(IPizzaRepository pizzas)
    : IRequestHandler<PlacePizzaOrder>
{
    public async Task HandleAsync(
        PlacePizzaOrder request,
        CancellationToken cancellationToken)
    {
        await pizzas.SaveAsync(
            new PizzaOrder(Guid.NewGuid(), request.CustomerId, request.Toppings, request.Address),
            cancellationToken);
    }
}

public class GetPizzaStatusHandler(IPizzaRepository pizzas)
    : IRequestHandler<GetPizzaStatus, PizzaStatusDto>
{
    public async Task<PizzaStatusDto> HandleAsync(
        GetPizzaStatus request,
        CancellationToken cancellationToken)
    {
        var order = await pizzas.FindAsync(request.PizzaId, cancellationToken);
        return new PizzaStatusDto(order.PizzaId, order.Status, order.EstimatedDelivery);
    }
}

Handlers are resolved from the DI container, so constructor injection works normally.

Dispatching

Inject IMediator wherever you need it:

public class PizzeriaService(IMediator mediator)
{
    public Task PlaceAsync(PlacePizzaOrderRequest req, CancellationToken ct)
        => mediator.SendAsync(new PlacePizzaOrder(req.CustomerId, req.Toppings, req.Address), ct);

    public Task<PizzaStatusDto> GetStatusAsync(Guid pizzaId, CancellationToken ct)
        => mediator.SendAsync<PizzaStatusDto>(new GetPizzaStatus(pizzaId), ct);
}

Sending to an unregistered request type throws InvalidOperationException at dispatch time.

Middleware

The mediator supports the same middleware levels as the rest of Emit. Register mediator-wide middleware on InboundPipeline, or per-handler middleware using the two-argument AddHandler overload:

emit.AddMediator(mediator =>
{
    // Runs for every request dispatched through this mediator
    mediator.InboundPipeline.Use<LoggingMiddleware>();

    // Runs only for PlacePizzaOrder requests
    mediator.AddHandler<PlacePizzaOrderHandler, PlacePizzaOrder>(handler =>
    {
        handler.Use<ValidationMiddleware>();
    });

    mediator.AddHandler<GetPizzaStatusHandler>();
});

The context flowing through all middleware layers is MediatorContext<TRequest>, which extends MessageContext<T>.

Observers

IMediatorObserver provides lifecycle hooks without writing middleware. An observer receives three callbacks: OnHandlingAsync (before the handler runs), OnHandledAsync (after successful completion), and OnHandleErrorAsync (on exception):

public class AuditObserver : IMediatorObserver
{
    public Task OnHandlingAsync<T>(MediatorContext<T> context)
    {
        return Task.CompletedTask;
    }

    public Task OnHandledAsync<T>(MediatorContext<T> context)
    {
        return Task.CompletedTask;
    }

    public Task OnHandleErrorAsync<T>(MediatorContext<T> context, Exception exception)
    {
        return Task.CompletedTask;
    }
}

Register observers on the mediator builder:

mediator.AddObserver<AuditObserver>();

Observer exceptions are swallowed and logged; a failing observer never interrupts the handler.

Transactional handlers

Decorate a mediator handler class with [Transactional] to run it in a transaction, which also makes it take part in the transactional outbox. Middleware registered on the handler runs inside the same transaction. To act once the transaction has committed, or once the request has definitely failed, register OnSucceeded and OnFailed callbacks on IEmitContext. See Unit of Work for transactions and callbacks, and the Transactional Outbox page for producing messages.

[Transactional]
public sealed class CreateOrderHandler(
    IEventProducer<string, OrderCreated> producer,
    MyDbContext db) : IRequestHandler<CreateOrderCommand>
{
    public async Task HandleAsync(CreateOrderCommand cmd, CancellationToken ct)
    {
        db.Orders.Add(cmd.Order);
        await producer.ProduceAsync(..., ct);
        // SaveChanges and transaction commit are handled by middleware.
    }
}

Streaming handlers

A handler that produces a sequence implements IStreamRequestHandler<TRequest, TResponse>, and its request implements IStreamRequest<TResponse>. Registration is unchanged.

public sealed record GetOrders(Guid CustomerId) : IStreamRequest<Order>;

public sealed class GetOrdersHandler(MyDbContext db) : IStreamRequestHandler<GetOrders, Order>
{
    public async IAsyncEnumerable<Order> HandleAsync(
        GetOrders request,
        [EnumeratorCancellation] CancellationToken ct = default)
    {
        await foreach (var order in db.Orders
            .Where(o => o.CustomerId == request.CustomerId)
            .AsAsyncEnumerable()
            .WithCancellation(ct))
        {
            yield return order;
        }
    }
}

Dispatch with CreateStreamAsync:

await foreach (var order in mediator.CreateStreamAsync(new GetOrders(id), ct))
{
    await response.WriteAsync(Serialize(order), ct);
}

A request declares one shape or the other. SendAsync will not accept a streaming request and CreateStreamAsync will not accept a single-response one, so the compiler picks for you.

What the pipeline does

The handler runs inside the middleware pipeline for the whole life of the stream, not just while the sequence is created. Middleware, observers and [Transactional] therefore wrap every item, and a duration metric measures the whole stream rather than its creation.

[Transactional] on a streaming handler holds one transaction open until the enumeration ends. That is what makes the stream atomic, and it is worth weighing for a long-lived connection, where it means holding a database transaction open for as long as the client stays connected.

Stopping early, and cancelling

The handler advances only when an item is requested, so it is always between items when you decide whether to continue.

Ending the enumeration, with a break or by taking a fixed number of items, stops the handler at its next item boundary and completes the request normally. Any work it already did is kept, and a transaction commits.

Cancelling the token aborts the request instead. The handler is cancelled where it stands and a transaction rolls back.

Disposing the enumerator completes the request, so if the request fails at that point, when a commit fails for instance, the failure is thrown from the loop.

Scope and lifetime

Handlers run in the scope that resolved the mediator, so the stream must be enumerated while that scope is alive. In ASP.NET that is the request scope, which lives as long as the response.

Nothing runs until you enumerate. Building a stream and never reading it starts no work.

One handler per request type

Registering two handlers for the same request type throws at startup. If you need fan-out, dispatch to multiple targets inside a single handler, or publish an event through the outbox.