Commands and Events on the Wire

Parts 3 and 4 of this series took apart Convey's in-memory CQRS spine: ICommand, IEvent, handlers discovered by assembly scan, dispatchers that create a DI scope per message. Parts 9 and 10 covered the RabbitMQ machinery: conventions, retries, dead letters. This part is about the thin seam that joins them — Convey.MessageBrokers.CQRS, which I think is the highest leverage-per-line package in the whole toolkit from the open-source DevMentors stack.

Here is the problem it solves. The Deliveries service needs to react when Orders publishes OrderCreated. Without help, that means writing a consumer: declare a queue, deserialize the payload, look up the right handler, create a scope, invoke, ack. Every team writes this class, and every team's version drifts. Convey's answer, visible in its own sample services, is one line in the pipeline:

app.UseRabbitMq()
   .SubscribeEvent<OrderCreated>()
   .SubscribeCommand<CreateDelivery>();

That's the entire consumer. No consumer class exists anywhere in the service.

Seventy lines, read in full

The package contains two files. The extension methods are short enough to quote in full — and worth quoting, because the design decision lives in them (Convey.MessageBrokers.CQRS/Extensions.cs):

public static IBusSubscriber SubscribeCommand<T>(this IBusSubscriber busSubscriber)
    where T : class, ICommand
    => busSubscriber.Subscribe<T>(async (serviceProvider, command, _) =>
    {
        using var scope = serviceProvider.CreateScope();
        await scope.ServiceProvider.GetRequiredService<ICommandHandler<T>>().HandleAsync(command);
    });

public static IBusSubscriber SubscribeEvent<T>(this IBusSubscriber busSubscriber)
    where T : class, IEvent
    => busSubscriber.Subscribe<T>(async (serviceProvider, @event, _) =>
    {
        using var scope = serviceProvider.CreateScope();
        await scope.ServiceProvider.GetRequiredService<IEventHandler<T>>().HandleAsync(@event);
    });

The generic IBusSubscriber.Subscribe<T> (from the broker abstractions package) takes a Func<IServiceProvider, T, object, Task>. These two methods close that gap with the same handler resolution the in-memory dispatchers use: fresh scope, GetRequiredService<ICommandHandler<T>>, invoke.

That is the design decision: a handler does not know whether its message arrived from the same process or from a broker. CreateDeliveryHandler : ICommandHandler<CreateDelivery> serves an HTTP-dispatched command today and a RabbitMQ-delivered one tomorrow, unchanged. When Orders outgrows synchronous calls to Deliveries, the migration is a publisher-side change; the handler and its tests never hear about it.

Publishing, and swapping the dispatcher

The publishing side is a pair of one-line aliases — busPublisher.SendAsync(command, messageContext) and busPublisher.PublishAsync(@event, messageContext) — over the generic IBusPublisher.PublishAsync. Semantically, commands and events are identical bytes on the wire; the distinction is entirely a naming discipline. What differs is the routing: Convey's conventions (part 9) derive exchange and routing key from the message type, and a contract class can pin them explicitly:

[Message("deliveries")]
public class DeliveryStarted : IEvent
{
    public Guid DeliveryId { get; }
    public DeliveryStarted(Guid deliveryId) => DeliveryId = deliveryId;
}

The quietly radical piece is ServiceBusMessageDispatcher, an internal class that implements both ICommandDispatcher and IEventDispatcher by forwarding to the bus:

public Task SendAsync<T>(T command, CancellationToken cancellationToken = default)
    where T : class, ICommand
    => _busPublisher.SendAsync(command, _accessor.CorrelationContext);

Register it with AddServiceBusCommandDispatcher() instead of AddInMemoryCommandDispatcher() and every _dispatcher.SendAsync(command) in your codebase silently becomes a broker publish. Location transparency in one registration.

I have used this exactly once in production, deliberately. It is a wonderful migration lever — you can peel a command's processing out of a service without touching call sites — and a terrible thing to leave ambient, because a call that used to be an awaited local operation with exceptions is now fire-and-forget with at-least-once delivery. The type system will not warn you that await SendAsync(...) no longer means “it happened.” If you flip this switch, flip it loudly: rename the registration in Program.cs, tell the team, and audit every caller that inspects state immediately after sending.

Correlation context: the invisible passenger

Notice the dispatcher passed _accessor.CorrelationContext as messageContext. This is Convey's mechanism for carrying request metadata — user id, trace id, origin — across service boundaries without any handler passing it along.

The plumbing has three stations. First, CorrelationContextAccessor in the abstractions package stores the context in an AsyncLocal, so it flows with the async call chain the way HttpContext does. Second, on publish, the RabbitMQ publisher serializes it into a message header (message_context by default, configurable in the rabbitmq options section). Third, on consume, the subscriber's background service reads that header and sets the accessor before invoking the handler — so when the handler (or anything it calls, including a further publish) asks for the correlation context, the one from the incoming message is already there.

The result: a context established when the HTTP request hit the Orders service is still attached to the third message in a chain three services away, and the logging package (part 19) can stamp every log line with it. Nobody wrote a parameter for it anywhere. The cost is the usual one for ambient state — it is typed as object end to end (ICorrelationContextAccessor.CorrelationContext { get; set; } is literally an object property), so producer and consumer must agree on its shape out-of-band, and nothing at compile time checks that they do.

One more subtlety worth knowing before an incident teaches it to you: the subscription lambdas discard the context parameter (_) and never flow a CancellationToken — HandleAsync(command) is called with the default token. Graceful shutdown will not cancel an in-flight broker-delivered handler the way it cancels an HTTP request; if your handler does long work, it needs its own timeout discipline (or the broker package's message-processing timeout from part 10).

When the handler throws: rejected events

Commands over a bus have an awkward property: the caller is gone. An HTTP dispatch returns 400 when validation fails; a bus dispatch returns nothing to nobody. Convey's convention for this is the rejected event. You register a mapper with the RabbitMQ package that translates exceptions into messages:

public class ExceptionToMessageMapper : IExceptionToMessageMapper
{
    public object Map(Exception exception, object message)
        => exception switch
        {
            OrderAlreadyExistsException ex when message is CreateOrder cmd
                => new CreateOrderRejected(cmd.OrderId, ex.Message, "order_already_exists"),
            _ => null
        };
}

When retries are exhausted (part 10 covered the Polly loop and dead-lettering), the consumer maps the exception and publishes the result as a new event. Whoever cares — typically an API gateway holding a WebSocket open to the browser — subscribes to CreateOrderRejected and closes the loop with the user. It is an honest pattern: it doesn't pretend async messaging can have synchronous error handling, it gives failure the same first-class transport as success. The discipline it demands is that every command you accept over the bus needs a thought-through rejection contract, or failures vanish into logs.

What the symmetry costs

I want to be fair about the trade-offs of “same handler, either transport,” because they bit me:

• Idempotency expectations differ silently. In-memory dispatch is exactly-once by construction. RabbitMQ is at-least-once, so the same handler must now tolerate redelivery. The interface cannot express this; your code review has to. (Part 12, on the outbox and inbox, is the systematic fix.)
• Contract coupling is by shared class. SubscribeEvent<OrderCreated>() needs the OrderCreated type, so services share contract classes — in Convey's samples, a copied Events/External class with a [Message] attribute pointing at the source exchange. Copying is the right call (shared contract assemblies become a coupling chokepoint), but it means renames are wire-breaking changes that the compiler cannot see across repos.
• Zero or many is legal for events, fatal for commands. GetRequiredService<ICommandHandler<T>> throws if no handler is registered — at message-processing time, not startup. A forgotten registration surfaces as a poison message in the dead-letter queue rather than a failed deploy.

None of these are flaws in the seventy lines; they are the actual physics of moving from calls to messages, surfacing exactly where they should. The package's virtue is that it adds nothing else — having read all of it, you know precisely what the wire does to your CQRS semantics, and what it can never do for you.

Next part: the strongest guarantee in the toolkit — the transactional outbox, and how Convey keeps your database and your broker telling the same story.