redb.Route.Amqp 4.1.0

redb.Route.Amqp

AMQP 1.0 transport for redb.Route via AMQPNetLite. Supports ActiveMQ Artemis, ActiveMQ Classic, Azure Service Bus, Amazon MQ, Qpid, and any AMQP 1.0 compliant broker.

Installation

dotnet add package redb.Route.Amqp

Usage

Fluent DSL

using redb.Route.Amqp.Fluent;

// Consume from AMQP address
From(Amqp.Address("orders")
        .Host("artemis.local").Port(5672)
        .User("admin").Password("secret")
        .Credit(10)
        .Durable())
    .Log("Received order: ${body}")
    .To("direct://process");

// Produce to AMQP address
From("direct://send")
    .To(Amqp.Address("events")
        .Host("artemis.local")
        .User("admin").Password("secret")
        .MessageDurable()
        .ContentType("application/json"));

// Azure Service Bus
From(Amqp.Address("orders")
        .Host("mybus.servicebus.windows.net").Port(5671)
        .User("policy-name").Password("shared-access-key")
        .Ssl()
        .ContainerId("route-consumer"))
    .To("direct://handle");

Fluent Builder API

Transactions

Inside a route's .Transacted() block a send joins the transaction: it goes out once the database has committed and is dropped if the block rolls back. .Transacted(false) sends at once, outside the transaction; .Transacted() requires an enclosing block and fails the step outside one. A request-reply producer (.ReplyTo()) always sends at once and refuses .Transacted().

The AMQP client would enlist a send in the ambient System.Transactions transaction by itself. The connector never lets it: a send that leaves at once runs with the ambient transaction suppressed and a deferred one after the transaction closed, so the broker never becomes a second resource next to the database. See the framework-wide Transactions guide (TRANSACTIONS.md in the redb.Route repository).

Several sends in one block: localTransactions

By default the sends a producer deferred in a block go out one after the other: a failure part-way leaves the earlier ones sent. With localTransactions=true (.LocalTransactions()) they commit in one AMQP local transaction: the connector opens a link to the broker's transaction coordinator, declares a transaction, sends every message carrying it and discharges it once — all of them arrive, or none. A send that cannot complete (the broker refuses it, or does not settle it within timeout seconds — Artemis withholds credit on a full address) discharges the transaction as failed, and the block's commit fails. The producer's transactions take turns on its link.

The broker must support AMQP local transactions: ActiveMQ Artemis, Apache Qpid, Azure Service Bus do; RabbitMQ's AMQP 1.0 plugin does not, and a broker that refuses the coordinator fails the commit with that said. The client's own System.Transactions support is not used: its discharge never got an answer from Artemis, while the same transaction driven through the protocol's own frames is answered at once.

Part of

redb.Route — ESB & EIP Framework for .NET

Named connection factory

Keep credentials out of the route URI: register a factory in the context registry and reference it by name. A set-but-unknown name fails loud at startup — a typo can never silently fall back to inline URI parameters.

context.AddToRegistry("prod", new AmqpConnectionFactory
{
    Host = "broker.internal",
    User = "svc-orders",
    Password = secrets.AmqpPassword,
});
// amqp://orders?connectionFactory=prod

Concurrency

The default is 1 concurrent consumer — the industry norm (Camel, Spring, the Azure SDK all ship 1): a single consumer preserves ordering and your handlers need no thread safety. Parallelism is an explicit opt-in:

concurrentConsumers=4       # a fixed worker count
concurrentConsumers=auto    # max(CPU count, 2) — the NServiceBus formula

Anything else — 0, a negative, a typo — fails at endpoint creation naming the option (the old int-typed option silently fell back to 1). Raising the value trades ordering for throughput: messages from the same queue are processed out of order, and your processors must be safe to run in parallel.