RabbitMQ by Convention

The first RabbitMQ code most .NET teams write is a page of ConnectionFactory, CreateModel, ExchangeDeclare, QueueDeclare, QueueBind — and the second copy of that page, in the next service, disagrees with the first about exchange durability. I have debugged that disagreement in production. Twice.

Convey's answer, in the open-source Convey.MessageBrokers.RabbitMQ package (part of DevMentors' toolkit — I'm a reader of this code, not its author), is to make topology a derivation rather than a declaration. You write:

services.AddConvey()
    .AddRabbitMq()
    .Build();

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

and everything else — connection management, exchange declaration, queue naming, binding, consuming — follows from configuration and conventions. Part 8 covered the abstractions; this part is the happy path through the RabbitMQ implementation. Part 10 is what happens when the happy path ends.

Two connections, opened before you're ready

The first surprise in Extensions.cs is that AddRabbitMq() opens two physical AMQP connections at registration time — one suffixed _producer, one _consumer — and registers them as singleton ProducerConnection and ConsumerConnection wrappers.

Separating them is textbook RabbitMQ practice: publishing and consuming on the same connection means a flood of outbound publishes can starve the TCP window that your acks travel on, and broker flow-control on publishers would throttle your consumers too. Most hand-rolled integrations never bother; Convey bakes it in, and that alone justifies reading this package.

The eagerness is the trade-off. The connections are created inside the AddRabbitMq() call itself — during dependency-injection registration, before the container even exists. If the broker is down when the process starts, the host never builds. There is no lazy reconnect-on-first-use; you either get fail-fast semantics (arguably what you want in Kubernetes, where a crash-looping pod is a visible signal) or a startup dependency you didn't ask for (definitely not what you want when the broker restarts during a deployment window). Automatic recovery after a successful start is delegated to the RabbitMQ client's networkRecoveryInterval, which the options expose.

The options section is the whole topology

Everything the package does is steered by the rabbitmq section. The sample Orders service ships a config that doubles as documentation:

"rabbitMq": {
  "connectionName": "orders-service",
  "messagesPersisted": true,
  "retries": 3,
  "retryInterval": 2,
  "conventionsCasing": "snakeCase",
  "hostnames": [ "localhost" ],
  "exchange": { "declare": true, "durable": true, "type": "topic", "name": "orders" },
  "queue": { "declare": true, "durable": true, "template": "{{assembly}}/{{exchange}}.{{message}}" },
  "context": { "enabled": true, "header": "message_context" },
  "deadLetter": { "enabled": true, "prefix": "dlx-" }
}

Note what is not here: no exchange-per-message declarations, no queue names, no bindings. Those are computed.

From class name to wire address

The conventions provider resolves three names per message type, cached per type:

• Exchange — the configured default (orders above) unless the type carries [Message(exchange: "...")].
• Routing key — the type name, run through the casing convention: OrderCreated becomes order_created under snakeCase.
• Queue — the template expanded: {{assembly}}/{{exchange}}.{{message}} yields Conveyor.Services.Deliveries/orders.order_created for the Deliveries service subscribing to Orders' event.

That queue template is quietly load-bearing. Because the consuming assembly's name is baked in, two different services subscribing to the same event get two different queues — which is exactly the fan-out-to-each-service, compete-within-a-service semantics you almost always want from a topic exchange. Nobody on the team has to know that; the template encodes it.

The cost of convention is a naming coupling I covered in part 8 and will keep repeating because it bites: rename the C# class and you have renamed the routing key and orphaned the old queue. And since the cache keys on Type but the derived name is just type.Name, two messages with the same class name in different namespaces collide silently.

Publishing: a channel per thread

Channels (IModel) are not thread-safe, and the naive fix — lock around one channel — serializes your throughput. Convey's RabbitMqClient instead keeps a channel pool keyed by managed thread id:

// Clients/RabbitMqClient.cs
channel = _connection.CreateModel();
_channels.TryAdd(threadId, channel);
_channelsCount++;

Each publishing thread gets a private channel, created on first use, reused forever after. It is simple and fast, with two sharp edges. First, channels are never released — on a thread pool that recycles thread ids this is effectively bounded, but the bound is enforced by a hard throw when MaxProducerChannels (default 1000) is hit, not by eviction. Second — and this is the one to internalize before you trust it with money — BasicPublish is called with no publisher confirms:

properties.Persistent = _persistMessages;
properties.MessageId = ...; properties.CorrelationId = ...;
channel.BasicPublish(conventions.Exchange, conventions.RoutingKey, properties, body.ToArray());

messagesPersisted: true marks the message durable once the broker has it, but the publish itself is fire-and-forget at the AMQP level. If the broker connection drops in the wrong few milliseconds, the publish vanishes and no exception tells you. For most event traffic that risk is acceptable; for the events you cannot lose, the answer is not confirms but the outbox — part 12.

What the publisher does do well is stamping: every message gets a MessageId, a CorrelationId (generated if absent), a Unix timestamp, the serialized correlation context under the message_context header, and — when tracing is on — the Jaeger span context under span_context. Parts 11 and 21 build on those headers.

Subscribing: a channel of subscriptions and one background service

SubscribeEvent<T>() looks imperative but is actually just a request. It writes a subscription record into an unbounded System.Threading.Channels channel (MessageSubscribersChannel); the actual consuming is done by RabbitMqBackgroundService, a hosted service that drains that channel and, for each subscription:

1. resolves exchange/routing key/queue from the conventions,
2. declares the exchange and queue (if declare: true) and binds them,
3. sets QoS — prefetch is forced to at least 1,
4. starts a consumer whose delivery callback deserializes the body to T and invokes your handler inside a fresh DI scope.

One consumer channel exists per {exchange}:{queue}:{routingKey} triple, acks are manual, and the handler signature is the same IEventHandler<T>/ICommandHandler<T> you met in part 3 — the broker is just another dispatcher. The topology therefore materializes lazily, at subscription time, from the same conventions the publisher used. The two sides never exchange a schema; they exchange a convention, and the broker's management UI is where you verify they agreed.

What I take from the happy path

Three things in this package are worth stealing even if you never run Convey:

1. Producer/consumer connection separation as a default, not an optimization someone remembers under load.
2. A queue template with the consuming assembly in it — one string that encodes correct pub/sub semantics for an entire fleet.
3. Uniform message stamping — id, correlation id, timestamp, context header — done in exactly one place, so every downstream feature (tracing, inbox dedup, logging) can rely on it.

And two things to go in with eyes open about: the eager connections make the broker a hard startup dependency, and the absence of publisher confirms means “persisted” is a promise about the broker's disk, not about the network in between.

The happy path ends where the handler throws. Retries, dead-letter exchanges, exception-to-message mapping and the plugin pipeline — the machinery that decides whether a poison message loops forever or lands somewhere you can inspect it — is part 10.