The RabbitMQ connector in redb.Route: the whole broker in one URI — RPC, competing consumers, dead-lettering. Still leaving MassTransit

We did Kafka earlier in this series. Now it's RabbitMQ's turn, and this one leans hard on how you actually use it. The redb.Route.RabbitMQ connector sits on top of the official RabbitMQ.Client 7.x, but you don't write a "client" — you write routes, and the whole broker collapses into a single URI:

From("rabbitmq://orders?exchange=shop&exchangeType=topic&routingKey=order.*&concurrentConsumers=4")
    .Log("Order in: ${body}")
    .To("direct://process");

Read it once and you already know the whole story: consume from queue orders, bound to the topic exchange shop on order.*, four consumers wide, log it, pass it on. No "open a channel," no "declare a binding," no "set the QoS" — the connector handles every bit of that.

About "leaving MassTransit," let me be straight with you: it's not that MassTransit can't do this — it can, RPC included (request/response via IRequestClient<T>). The difference is the model. MassTransit is a bus: message contracts, consumers, bus configuration — you buy into its whole worldview at once. redb.Route is explicit routes in the Apache Camel spirit: an endpoint is a URI, integration patterns are steps in a route, the transport is an abstraction underneath. Fewer ceremonies around contracts and registrations, more plain "from here → to there → and here's what happens along the way." Which one fits you is a matter of taste and of the job in front of you; this post is about what RabbitMQ looks like in the second model.

Here's the map: the anatomy of the URI, every parameter as a reference table (what it does and why), the properties the consumer and producer move to and from headers, competing consumers (concurrency), broadcasting across cluster nodes (a real production example), RPC (RabbitMQ makes this one especially clean), and the two EIP patterns that fit a broker like a glove — Filter (on compiled predicates) and Dead Letter Channel.

All the code is in English; the examples aren't invented — they live in redb.Route/demos/redb.Route.Demo/Routes, the same demo project that runs in CI.

Part of the redb / redb.Route series — recent posts first:

• Leaving MassTransit for a Camel state of mind: the Kafka connector, Scatter-Gather, and transactions
• Apache Camel for .NET, dissected: the HTTP connector with no ASP.NET MVC + the Content-Based Router
• redb.Route — Apache Camel for .NET: 22 transports, 30+ EIP patterns, compiled DSL

Sources: github.com/redbase-app. About the database itself: redb.ru.

---

Anatomy of the URI

An endpoint is a string:

rabbitmq://queue-name?host=localhost&exchange=my-exchange&exchangeType=topic&routingKey=order.*

Scheme rabbitmq://, the URI host is the queue name, then query parameters (names are in the table below). The same string works as a consumer (in From(...)) and as a producer (in .To(...)) — the role is decided by where the endpoint sits in the route. You don't have to repeat the connection: register a connection factory once and point at it with connectionFactory=name — host, credentials and SSL live in one place, and the connection gets reused.

One detail worth pausing on: parameter values are templates. routingKey=order.new is a constant, but routingKey=order.${header.type} is dynamic — the key is computed per message from its header (the same mechanism as Kafka's partition key). So "a string" here doesn't mean "static only" — a ${...} inside the value turns it into an expression.

On the fluent builder. The connector also ships a builder, Rabbit.Queue("orders")…, but it's expression-first: the string-ish methods take an expression so a producer value can be computed per message (.RoutingKey(Header("order.type")) — a dynamic key, Kafka-style). For a plain constant there's now a string overload too — .Host("localhost") and .RoutingKey("order.new") just work, and ${...} still interpolates. In this post I use the URI form — for static config it's shorter and easier to read at a glance; the dynamic-via-expressions side gets its own writeup.

---

Every parameter

The reason to bookmark this post. Names are exactly as they appear in the URI; defaults are what you get out of the box.

Connection

Parameter	Default	What it's for
host	localhost	Broker host
port	5672	AMQP port
username / password	guest / guest	Credentials
virtualHost	/	Virtual host — resource isolation inside the broker
connectionFactory	—	Name of a factory in the registry: settings in one place, connection reused
clientName	redb.Route	Connection name — shows up in the management UI

Exchange

Parameter	Default	What it's for
exchange	(empty)	Exchange name; empty = the default exchange (routed by queue name)
exchangeType	direct	direct / topic / fanout / headers
exchangeDurable	true	Survives a broker restart
exchangeAutoDelete	false	Delete once it's no longer in use
declare	false	Create the exchange + queue on startup if they don't exist

Queue

Parameter	Default	What it's for
queue	(from the URI host)	Name; empty = an auto-generated exclusive queue
durable	true	Survives a restart
autoDelete	false	Delete once the last consumer leaves
exclusive	false	Only one consumer allowed

Routing and message

Parameter	Default	What it's for
routingKey	(empty)	Key for publish and bind
contentType	application/json	Message content type
messageTtl	0	Message TTL, ms (x-message-ttl)
expires	0	Queue is deleted after N ms with no consumers (x-expires)

Consumer

Parameter	Default	What it's for
concurrentConsumers	1	The one and only concurrency knob — up to N messages handled at once
prefetchCount	10	Prefetch per consumer; keep it ≥ concurrentConsumers
autoAck	false	true = at-most-once (no requeue on error); false = at-least-once
transacted	false	Transactional AMQP channel (mutually exclusive with autoAck)
mandatory	(auto)	Return an undeliverable message; defaults to true for direct/headers

RPC

Parameter	Default	What it's for
replyTo	false	Turn on request/reply: publish the request and wait for the correlated response
timeout	60	How long to wait for the reply, seconds
maxOutstandingConfirms	2048	Cap on unconfirmed publishes — so you don't trip over the client's rate limiter under load

Queue limits and dead-lettering

Parameter	Default	What it's for
maxLength / maxLengthBytes	0	Queue cap in messages / bytes (x-max-length[-bytes])
overflow	—	On overflow: drop-head / reject-publish / reject-publish-dlx
deadLetterExchange	—	DLX — where dead messages go (x-dead-letter-exchange)
deadLetterRoutingKey	—	Routing key used when dead-lettering
queueType	—	classic / quorum (a replicated queue for a cluster)
maxPriority	0	Max priority for priority queues

Resilience and TLS

Parameter	Default	What it's for
automaticRecovery	true	Auto-recover the connection
topologyRecoveryEnabled	true	Re-declare exchanges/queues/binds after recovery
recoveryInterval	5	Retry interval for recovery, seconds
heartbeat	60	Heartbeat interval, seconds — broker and client ping each other so a dead connection gets noticed in time
connectionTimeout	60	Don't hang on an unreachable node
ssl	false	Turn on TLS
sslServerName / sslCertPath / sslCertPassphrase	—	SNI and a client cert for mTLS

The startup validator won't let you shoot yourself in the foot: prefetch/concurrentConsumers/timeout must be > 0, and autoAck + transacted together is flat-out rejected.

---

AMQP properties ↔ headers: a full round-trip

The connector maps a RabbitMQ message's properties to exchange headers both ways — on the consumer and on the producer. When the consumer takes a delivery, it spreads BasicProperties across the headers; when the producer sends, it reads them back and puts them on the outgoing message. So a consume → produce hop carries every AMQP property straight through, and you can set or override any of them by just placing a header.

Naming comes in two tiers.

Delivery metadata — there's no standard AMQP name for these, so they get the redbRmq. prefix:

Header	What's in it
redbRmq.Exchange	the exchange the message arrived on
redbRmq.RoutingKey	the delivery's routing key
redbRmq.Redelivered	redelivery flag (after a nack/requeue or a consumer crash)
redbRmq.DeliveryTag	delivery tag (used for the ack)
redbRmq.ConsumerTag	consumer tag

Standard message properties — under their well-known names, no prefix (everyone knows them already, and the round-trip carries them as-is):

Header	AMQP property
CorrelationId	correlation-id
ReplyTo	reply-to — the address to reply to
MessageId	message-id
Priority	priority 0–9
Expiration	message TTL, ms
Type / AppId / UserId	type / app / user
ContentType / ContentEncoding	content-type / encoding
Timestamp	timestamp
Persistent	persistent / transient

redbRmq.Redelivered earns its keep: use it to handle "this is already the second attempt" separately — send the redelivery down a different route instead of looping forever on requeue.

The ReplyTo header ≠ the replyTo parameter. Similar names, different things. The URI parameter replyTo=true switches the client into RPC mode (below). The ReplyTo header is the AMQP reply-to property itself: set it on a normal send and you're pointing the message's reply address directly, and the producer will carry the header into the property. All of this works because the mapping is two-way — the consumer doesn't just read these; the producer honors them too.

---

Recipe 1: pub/sub

A fanout exchange fans a copy out to every subscriber; the key is ignored. The producer publishes to the exchange, each consumer declares its own queue and binds:

// publisher — publishes to a fanout exchange (a copy to every subscriber)
From("timer://tick?period=5000")
    .SetBody(_ => BuildEvent())
    .To("rabbitmq://events-pub?exchange=events&exchangeType=fanout");

// two independent subscribers — each with its own queue
From("rabbitmq://audit?exchange=events&exchangeType=fanout&declare=true").To("direct://audit");
From("rabbitmq://email?exchange=events&exchangeType=fanout&declare=true").To("direct://email");

Want smarter routing? Switch to exchangeType=topic and a routingKey with wildcards (order.*, #.error). Only the exchange type and the key change.

Recipe 2: competing consumers (concurrency)

One parameter — concurrentConsumers:

From("rabbitmq://orders?concurrentConsumers=8&prefetchCount=16")   // up to 8 in flight
    .Process(HandleOrder);

This is the single knob for consumer concurrency: it sets both the channel's AMQP dispatch concurrency and the app-level semaphore. The default is 1 (strictly serial, order preserved). Set N > 1 and per-queue ordering is no longer guaranteed, your handler has to be thread-safe, and you keep prefetchCount ≥ N.

A note for anyone who kicked the tires on an earlier build: before 3.2.2, concurrentConsumers quietly gave you no parallelism thanks to a trap in RabbitMQ.Client 7.x (the channel was pinned to serial dispatch). Since 3.2.2 it's the one knob and it actually works — if you set N > 1 and "it somehow didn't speed up," update.

Acknowledgements are a parameter too. The default is at-least-once: ack after a clean pass, nack + requeue on error. Want fire-and-forget? autoAck=true flips you to at-most-once (the broker settles the delivery on dispatch; an error doesn't bring the message back).

And if what you need to parallelize is the processing inside a route rather than the intake (a serial source but heavy work), there's an orthogonal EIP step, .Threads(N). concurrentConsumers scales reading from the queue; .Threads scales the work; they compose freely.

Recipe 3: broadcasting across cluster nodes (local-cache invalidation)

A pattern straight from production. A service runs as a cluster of several nodes, each with its own local cache. Write to the database on any node, and you want the cache to refresh on all of them. The classic move over RabbitMQ: an audit producer publishes a domain event to a durable topic exchange, and every node hangs its own temporary queue off it and listens.

Producer (wherever the DB write happens) — a route per event kind, all into one exchange lt.events, each with its own routing key. Stamp the headers up front so subscribers can route/filter without cracking open the body. The connection comes from a shared factory:

// shared stamp: domain headers for subscribers — routing without parsing the body
Action<IExchange> stamp = e =>
{
    var a = e.In.GetBody<AuditEvent>();
    e.In.Headers["CorrelationId"] = a.CorrelationId;
    e.In.Headers["EntityType"]    = a.EntityType;
    e.In.Headers["EventType"]     = a.EventType.ToString();
};

From("direct://audit.entity.created")
    .RouteId("audit-entity-created")
    .Process(stamp)
    .SetBody(e => JsonSerializer.Serialize(e.In.GetBody<AuditEvent>()))
    .To("rabbitmq://?connectionFactory=RabbitMQ&exchange=lt.events&exchangeType=topic"
        + "&routingKey=dal.events.created&exchangeDurable=true");

// updated / deleted / batch / error — same exchange, routingKey=dal.events.{updated|deleted|…}

A consumer on every node — here's the whole trick. queue= empty + autoDelete=true: each node gets its own server-named temporary queue bound to the shared exchange. A node disconnects, its queue is gone. So a copy of every event lands on every node:

// created and updated — two consumers into one shared handler
From("rabbitmq://?connectionFactory=RabbitMQ&exchange=lt.events&exchangeType=topic"
        + "&routingKey=dal.events.created&queue=&autoDelete=true&expires=300000"
        + "&concurrentConsumers=2&prefetchCount=10")
    .RouteId("cache-consumer-created")
    .To("direct://cache-update");

From("rabbitmq://?connectionFactory=RabbitMQ&exchange=lt.events&exchangeType=topic"
        + "&routingKey=dal.events.updated&queue=&autoDelete=true&expires=300000"
        + "&concurrentConsumers=2&prefetchCount=10")
    .RouteId("cache-consumer-updated")
    .To("direct://cache-update");

// one processor for both streams — refreshes this node's local cache
From("direct://cache-update")
    .RouteId("cache-update")
    .Process(new CacheUpdateProcessor(/* … */));

Here's the key bit — the contrast with Recipe 2. That was competing consumers: one shared durable queue, the work split across consumers (each message handled by exactly one of them). This is the opposite: every node has its own ephemeral queue on the shared exchange → a broadcast, every node sees every event. Same connector, opposite topology, and the whole difference is a single parameter (queue=name&durable=true versus queue=&autoDelete=true). A DB write on any node → one event to the exchange → the local cache refreshes on every node, with no polling and no single point of failure.

Recipe 4: RPC (request/reply)

This is where it gets genuinely nice. Classic RabbitMQ RPC means a temporary reply queue, a correlationId, and waiting for the answer. Here it's one parameter on the client and nothing on the server. Here's how the demo wires it (DemoEndpoints.cs):

// server — an ordinary consumer; declare creates the queue
public const string RabbitConsumer =
    "rabbitmq://demo-rpc-queue?host=localhost&port=5672&username=admin&password=admin&declare=true";

// client — replyTo=true turns on RPC, timeout=15 — wait no longer than 15s
public const string RabbitProducer =
    "rabbitmq://demo-rpc-queue?host=localhost&port=5672&username=admin&password=admin&replyTo=true&timeout=15";

The client is a plain .To(...), except it doesn't fire and forget — it waits for the reply and rolls on carrying it (MainPipelineRoutes.cs):

.Log("[3-RABBIT] → Sending to RabbitMQ RPC...")
.To(RabbitProducer)                                      // publish the request and WAIT for the reply
.Log("[3-RABBIT] ← stamp.rabbit=${header.stamp.rabbit}") // the reply is already on the exchange

The server is just a consumer — there's nothing to configure for RPC. The connector sees reply-to on the incoming message, makes the route InOut, and sends whatever's in Out back to the reply queue with the same correlationId. The entire "server" from the demo:

From(RabbitConsumer)
    .RouteId("demo-rabbit-worker")
    .Log("[RABBIT-W] ▶ Received: ${body}")
    .SetHeader("stamp.rabbit", e => $"ok:{DateTime.UtcNow:HH:mm:ss.fff}")
    .Log("[RABBIT-W] ◀ Stamped, replying");         // the reply goes out on its own

Not a word about reply queues or correlation — the client sent a request and got back the stamp.rabbit header the server stamped on. Clean request/reply over a broker, with none of the temporary-queue plumbing by hand.

This is exactly where the contrast with MassTransit shows. It has request/response too — but that's IRequestClient<TRequest>, typed TRequest/TResponse contracts, and a consumer registered on the bus. Powerful, but it's a separate abstraction you have to step into. Here it's replyTo=true in the URI, and on the server side, nothing at all: an ordinary consumer, and the connector saw reply-to and replied. Two takes on the same problem: contracts and a bus versus one flag on an endpoint.

Recipe 5 (EIP): Filter — let only what matters through

Message Filter is a plain, perpetually useful EIP: let through only the messages that satisfy a condition and quietly drop the rest. In redb.Route that's .Filter(...) … .EndFilter() (or the shorter .End() — same thing): the block between them runs only for messages that pass, and for everything else the route ends right there. And this is where the compiled predicate engine shows up — the same one we got into in the expressions piece.

The condition can be written three ways. As a predicate over a header (or the body):

From("rabbitmq://orders?concurrentConsumers=4")
    .Filter(Header("score").isGreaterThan(50))       // score > 50
        .To("rabbitmq://vip-orders")
    .EndFilter();

The predicate set is everything you'd expect: .isEqualTo / .isNotEqualTo / .isGreaterThan / .isLessThan / .isGreaterThanOrEqualTo / .isLessThanOrEqualTo / .isBetween(a, b) / .contains(...). On the left, Header("...") or Body():

.Filter(Header("email").contains("@example"))   .To("direct://internal") .EndFilter();
.Filter(Header("grade").isEqualTo("B+"))         .To("direct://grade-b")  .EndFilter();

As a string expression — the ${...} resolves and the result is coerced to a boolean. For a truthy check, just the value; for a comparison, wrap it in logical(...) (a bare ${header.score > 50} inside a template comes back as an empty string — you can't write it that way):

.Filter("${header.status}")                  // 'active' → truthy → passes
    .Log("active only")
.EndFilter();

.Filter("${logical(header.score > 50)}")     // comparison → "True"/"False"
    .To("rabbitmq://vip")
.EndFilter();

Or as a lambda, when you want arbitrary C#:

.Filter(ex => ex.In.Body is string s && s.Contains("urgent"))
    .To("rabbitmq://urgent")
.EndFilter();

The handy thing is filtering on the very headers the consumer already stamped — say .Filter(Header("redbRmq.RoutingKey").contains("order")), so only messages with the routing key you care about make it into the block.

How an expression becomes a bool

Under the hood, for each message Filter evaluates the expression to object? and coerces it to bool by the ConvertToBoolean rules:

bool   → as-is
string → "true"/"false" parse; otherwise a non-empty string = true, empty = false
null   → false
anything else (number/object, non-null) → true

Which explains how the three forms behave:

• A predicate like Header("amount").isGreaterThan(1000) is an IPredicate with a ready Func<IExchange,bool>. No coercion needed: a real boolean straight from the comparison. It compiles to a delegate and gets cached — not reflection per message, but a fast call.
• A template string like "${header.status}" — the ${...} interpolates to a value, then the truthy rules apply. status="active" → non-empty string → true (that's why the demo says "active is truthy"); and "${header.enabled}" with "false" → bool.TryParse → false. A comparison written straight into the template ("${header.score > 50}") doesn't count — it comes back empty (i.e. false); for a comparison you need logical(...), as above.
• A lambda ex => … — whatever you return is the predicate.

So "a boolean expression" is either an honest bool from a predicate/comparison, or a truthy coercion of an interpolated value. An empty string, null and "false" cut it off; a non-empty string, a number, true let it through.

The DSL is open: your own step is an extension method

Worth noticing one thing here. The string and predicate forms Filter(string) / Filter(IPredicate) aren't on the IRouteDefinition interface — that has only the canonical Filter(Func<IExchange,bool>) and Filter(IExpression). The string form is an extension method, a thin façade over the canonical one:

// the whole string overload (redb.Route/…/RouteDefinitionCamelDslExtensions.cs):
public static FilterDefinition Filter(this IRouteDefinition self, string simpleTemplate)
    => self.Filter(new StringExpression(simpleTemplate));   // wrap the string in a template expression

Which carries an important corollary: the DSL is extensible, and a step of your own is added the exact same way — an extension method that delegates to the canonical one. Want a custom "verb" in your routes? Write:

// your own step: pass only "urgent" in the body
public static FilterDefinition OnlyUrgent(this IRouteDefinition self)
    => self.Filter(ex => ex.In.Body is string s && s.Contains("urgent"));

// and in a route it reads like a native step:
From("rabbitmq://orders")
    .OnlyUrgent()
        .To("rabbitmq://urgent")
    .EndFilter();

No magic, no engine edits — just an extension method over IRouteDefinition. That's how the framework itself does the Camel-compatible forms, the string templates, and the predicate overloads. So if a step is missing for you, you write it yourself — no forking the connector.

We've only brushed the edge of expressions here — Header(...), Body(), ${...}, logical(...). Expressions in redb.Route are really a small compiled language of their own: access to the body and headers, exchange properties, indexers (items[0]), string functions, JSONPath and XPath. All of it gets tokenized, parsed into an AST, and compiled to a delegate (System.Linq.Expressions, not reflection), with a cache. It's a big topic — the engine gets its own post; it's earned it.

Recipe 6 (EIP): Dead Letter Channel

Of the Hohpe & Woolf patterns, Dead Letter Channel fits RabbitMQ perfectly — because on the broker it's a built-in feature (x-dead-letter-exchange), not something emulated in code. The idea: a message that didn't process (or expired on TTL, or didn't fit the queue's limit) isn't lost and doesn't spin forever on requeue — it heads to the morgue, a separate dead-letter exchange you can pick apart at your leisure.

It's set right in the working queue's URI:

// working queue: on nack-without-requeue / expiry / overflow the message goes to the DLX
From("rabbitmq://orders?declare=true"
        + "&deadLetterExchange=orders.dlx"          // where
        + "&deadLetterRoutingKey=orders.failed"     // with which key
        + "&messageTtl=30000"                       // not processed in 30s → DLX
        + "&overflow=reject-publish-dlx&maxLength=100000"   // and overflow goes there too
        + "&concurrentConsumers=4")
    .Process(HandleOrder);

// the morgue: a separate route listens on the DLX — logs and stores into redb for triage
From("rabbitmq://orders.dead?declare=true&exchange=orders.dlx&exchangeType=direct&routingKey=orders.failed")
    .Log("[DLQ] dead-lettered: ${body}")
    .ProcessWithRedb((redb, ex) => redb.SaveAsync(BuildDeadLetter(ex)));

No hand-rolled retry loop, no "sleep and try again" — reliability is handed to the broker, and what's left on your side is a clean triage route that writes the problem messages into a database. And the queue doesn't balloon under a surge: the excess leaves for the DLX right away instead of piling up in the broker's memory.

---

Wrap-up

RabbitMQ in redb.Route is routes where the whole broker is one URI: From("rabbitmq://…") / .To("rabbitmq://…"), with all the channel juggling, QoS, binds, reply queues and dead-lettering tucked behind the string's parameters. Every exchange type, durable/quorum queues, priorities, TTL, DLX, auto-recovery for a cluster, TLS/mTLS — the set is complete, and what you end up with is a short, legible string. Competing consumers turn on with one parameter, RPC with one flag on the client and zero config on the server, Filter trims the noise with compiled predicates, and Dead Letter Channel is a couple of parameters over the native DLX.

The full demo is in redb.Route/demos/redb.Route.Demo — it comes up against a local RabbitMQ (admin/admin) and runs as a self-test. If something bites in your scenario, say so in the comments.

Sources and releases: github.com/redbase-app. About the redb database: redb.ru.

If this was useful — a ⭐ on GitHub helps others find it.