Exchanges, queues and bindings

Queues

if err := mq.DeclareQueue(ctx, "orders"); err != nil {
    return err
}

Durable by default, and — because it is durable — a quorum queue. Options:

err := mq.DeclareQueue(ctx, "orders",
    acemq.DeadLetterTo("orders-dead"),
    acemq.QueueArg("x-max-length", 100000))

Quorum, classic and streams

orders above is declared with x-queue-type: quorum, and so is any durable queue Topology.Queue(...) adds. The type is part of the queue's identity as far as the broker is concerned, so this is a cross-language contract rather than a preference: two services consuming orders both declare orders, and the second one to disagree about the type is refused with PRECONDITION_FAILED and cannot consume at all. Java has declared source queues quorum since before the other libraries existed, and a quorum queue cannot be redeclared as anything else, so quorum is what all five agree on.

Three kinds of queue are left classic:

• Anything exclusive, auto-deleting or transient. RabbitMQ allows a quorum queue to be none of the three, so a default that applied to them would not be a slower queue but a declaration the broker rejects — at start-up, wherever a reply queue, a health probe or a temporary queue is created.
• The queues underneath a source queue: every {queue}.retry.{delay} rung, {queue}.dlq and {queue}.parked. Nothing consumes a rung, and the two dead-letter queues are drained by a person. Java declares all of them QueueType.CLASSIC and this library matches.
• Anything that named its own type, with acemq.OfType or by writing x-queue-type with acemq.QueueArg.

acemq.OfType(acemq.QueueClassic) sends no x-queue-type argument, because that is what classic is on the wire in Java, .NET, Python and Ruby. Asking for classic and inheriting classic therefore produce the identical argument table, which matters wherever two tables are compared rather than two brokers consulted — the in-memory transport here, and anybody diffing two topologies.

A queue that is already classic on the broker is not converted. Its type cannot be changed after it is created, so the declaration is refused; drain the queue and recreate it, or declare it acemq.OfType(acemq.QueueClassic).

Redeclaring

Declaring is idempotent while the settings match. When they do not, RabbitMQ refuses:

acemq: cannot declare queue "orders": Exception (406) Reason: "PRECONDITION_FAILED
- inequivalent arg 'durable' for queue 'orders'"

That refusal is passed straight on rather than swallowed. It means the code and the broker disagree about what the queue is — usually because somebody changed the declaration without migrating the queue — and carrying on regardless would leave a service running against a queue it does not think it has.

It also kills the channel it happened on, which is why the error surfaces immediately rather than at the next publish.

Exchanges

if err := mq.DeclareExchange(ctx, "events", "topic"); err != nil {
    return err
}

Durable by default; acemq.TransientExchange() and acemq.ExchangeArg(...) adjust it. The kind is direct, topic, fanout or headers.

Bindings

if err := mq.Bind(ctx, "eu-orders", "events", "orders.*.eu"); err != nil {
    return err
}

Both the queue and the exchange must already be declared.

The default exchange

Publishing with an empty exchange name uses the default exchange, which routes to the queue whose name matches the routing key. No exchange, no binding:

pub := acemq.NewPublisher[OrderPlaced](mq, "", "orders")

Right for a work queue. Wrong as soon as a second consumer wants the same messages, because it can only ever reach one queue.

Topic patterns

* is exactly one word. # is zero or more.

Note orders.# matching the bare word orders: a trailing # matches nothing at all, which is the case most implementations get wrong.

The in-memory transport matches word by word rather than by translating the pattern into a regular expression. The translation looks shorter and is wrong on exactly the patterns nobody writes until they do — a trailing #, a # beside a *, a bare #. Twenty-six cases pin it, and a test against a real broker checks that RabbitMQ agrees; if the two disagreed, the in-memory transport would be certifying behaviour that does not happen.

Fanout

Every bound queue gets a copy and the routing key is ignored:

if err := mq.DeclareExchange(ctx, "broadcast", "fanout"); err != nil {
    return err
}
for _, q := range []string{"audit", "search-index"} {
    if err := mq.DeclareQueue(ctx, q); err != nil {
        return err
    }
    if err := mq.Bind(ctx, q, "broadcast", ""); err != nil {
        return err
    }
}

A shape that usually works

One topic exchange per bounded context, one queue per consuming service, bound to the patterns it cares about:

if err := mq.DeclareExchange(ctx, "orders-events", "topic"); err != nil {
    return err
}

if err := mq.DeclareQueue(ctx, "shipping-orders",
    acemq.DeadLetterTo("shipping-dead")); err != nil {
    return err
}
if err := mq.Bind(ctx, "shipping-orders", "orders-events", "order.placed"); err != nil {
    return err
}
if err := mq.Bind(ctx, "shipping-orders", "orders-events", "order.cancelled"); err != nil {
    return err
}

The publisher names an event, not a destination. A new consumer is a new queue and a new binding, and the publisher never learns about it.

Declaring it all at once

Call-at-a-time works, and stops working the moment somebody needs to know what a service will do to a broker before it does it.

topology := acemq.NewTopology().
    Exchange("orders-events", "topic").
    Queue("shipping-orders", acemq.DeadLetterTo("shipping-dead")).
    Queue("shipping-dead").
    Binding("shipping-orders", "orders-events", "order.placed").
    Binding("shipping-orders", "orders-events", "order.cancelled")

if err := topology.Apply(ctx, mq); err != nil {
    return err
}

Exchanges, then queues, then bindings — the order a broker needs.

The queues a consumer needs to fail into

A consumer that gives up on a message republishes it to {queue}.dlq with the reason attached, one that cannot decode a body republishes it to {queue}.parked, and one whose policy has long waits republishes into a {queue}.retry.{delay} rung. All three have to exist, or the publish is refused and the message goes back to the broker with nothing recorded about why:

policy := acemq.ExponentialRetry(6, 10*time.Second, 0)

topology := acemq.NewTopology().
    Queue("orders").
    DeadLetters("orders").     // orders.dlq, orders.parked, acemq.dlx and the wiring
    Retries("orders", policy)  // orders.retry.40s, .80s, .160s and their way home

acemq.Consume declares all of them itself when it starts, from the policy it was given, with the same arguments this plan uses. The two agree in either order, so a service that applies its topology and then starts a consumer is not refused and neither is a consumer starting against a broker somebody else set up. Write the plan anyway — it is what a person reviews, and it carries the source queue's own dead-letter arguments, which no consumer declares — but the failure it used to be the only guard against is gone: a publish into a queue nobody declared is unroutable, and the broker discards an unroutable message without a trace.

Printing it is the point — this is what topology.String() gives:

Topology: 2 exchanges, 6 queues, 3 bindings
  declare exchange acemq.dlx (direct)
  declare exchange acemq.retry (direct)
  declare queue orders (durable, quorum, x-dead-letter-exchange=acemq.dlx, x-dead-letter-routing-key=orders.dlq)
  declare queue orders.dlq (durable, classic)
  declare queue orders.parked (durable, classic)
  declare queue orders.retry.40s (durable, classic, x-dead-letter-exchange=acemq.retry, x-dead-letter-routing-key=orders, x-message-ttl=40000)
  declare queue orders.retry.80s (durable, classic, x-dead-letter-exchange=acemq.retry, x-dead-letter-routing-key=orders, x-message-ttl=80000)
  declare queue orders.retry.160s (durable, classic, x-dead-letter-exchange=acemq.retry, x-dead-letter-routing-key=orders, x-message-ttl=160000)
  declare binding orders.dlq (from acemq.dlx on orders.dlq)
  declare binding orders.parked (from acemq.dlx on orders.parked)
  declare binding orders (from acemq.retry on orders)

DeadLetters puts two arguments on orders itself, and they are the reason it has to declare the queue rather than only its dead letters: they are part of that declaration, and every other AceMQ library writes the same two. Two services consuming orders both declare it, so a disagreement here is a PRECONDITION_FAILED for whichever started second. Asking for dead letters on a queue this topology does not declare is refused for the same reason — there would be nowhere to put them.

Both acemq.dlx and acemq.retry are shared. However many queues use them, each appears once in the plan, and neither is deleted by anything: they belong to the broker rather than to a service.

Retries takes the policy rather than a list of delays because the rungs a consumer publishes into are derived from the policy it runs. A second list would be free to drift from the first, and the way that drift shows up is a retry published into a queue nobody declared, at the moment the service is already failing. See reliability for what a rung is and why the waiting happens there.

It is checked before the broker sees it

A binding naming a queue the topology does not declare is refused here rather than by the broker:

acemq: binding orders-events -> a-queue-nobody-declared (order.placed) names
queue "a-queue-nobody-declared", which this topology does not declare

The broker would have accepted it, whenever that queue happened to exist already — and the service would then depend on something nothing declares, until a fresh environment where it fails at start-up for reasons nobody can see.

Reading it before applying it

fmt.Println(topology)

Topology: 1 exchanges, 2 queues, 2 bindings
  declare exchange orders-events (topic)
  declare queue shipping-orders (durable, quorum, x-dead-letter-exchange=shipping-dead)
  declare queue shipping-dead (durable, quorum)
  declare binding shipping-orders (from orders-events on order.placed)
  declare binding shipping-orders (from orders-events on order.cancelled)

A deployment that changes a broker should be something somebody can read first. This is a statement of intent, not a diff against the live broker — AMQP cannot enumerate what is there without the management API, and a plan that quietly guessed would be worse than one honest about what it is.

Reading it against the broker it will change

ApplyWith takes a mode. DryRun asks the broker about each queue and changes nothing:

plan, err := topology.ApplyWith(ctx, mq, acemq.DryRun)
for _, action := range plan {
    log.Println(action)
}

exchange orders-events: topic — unknown, AMQP cannot report exchanges
queue shipping-orders: durable, quorum, x-dead-letter-exchange=shipping-dead — differs:
  the broker refused the declaration: Exception (406) PRECONDITION_FAILED
queue shipping-dead: durable, quorum — would create
binding shipping-orders: from orders-events on order.placed — unknown, AMQP
  cannot report bindings

Three things are worth knowing about that output:

• Queues are asked about, not declared. A passive declare answers NOT_FOUND for a queue that is not there rather than creating it, so a dry run never creates what it describes. There is a test against a real broker that the queue is still missing afterwards.
• Exchanges and bindings say "unknown". AMQP has no way to read them back, and "would create" about something that already exists is the kind of plausible-looking output that stops being read.
• Declare is the ordinary mode, and Apply is the same thing without the plan.

A transport that cannot be asked — a custom one that implements neither DriftChecker nor QueueInspector — reports every queue as unknown rather than pretending.

Drift

A service and its broker disagreeing about a queue is the failure that shows up as messages going somewhere nobody is looking: a dead-letter exchange that was changed, a durability that was not.

reports, err := topology.Check(ctx, mq)
for _, r := range reports {
    log.Printf("drift: %s", r)
}

drift: queue shipping-orders: the broker refused the declaration:
Exception (406) PRECONDITION_FAILED - inequivalent arg 'x-dead-letter-exchange'

PRECONDITION_FAILED is the only way AMQP will report this without the management API. Two consequences worth knowing:

• A failed declaration kills its channel, so the check uses a channel of its own and throws it away. Doing it on the shared publishing channel would take the connection's publishing down with a question. There is a test that the connection still works afterwards.
• A queue that is not there is passed over, not declared. A queue that does not exist cannot disagree with anything, and a check that created one as a side effect of asking would not be safe to run against a broker somebody is only inspecting.

Asking about a queue, and removing one

count, err := mq.MessageCount(ctx, "shipping-orders")
exists, err := mq.QueueExists(ctx, "shipping-orders")
err := mq.DeleteQueue(ctx, "shipping-orders")

MessageCount is a number for a dashboard or a test, not a decision to make in a handler: it is a snapshot of a queue that is still moving, and by the time it is read somewhere else it is already wrong.

QueueExists creates nothing — it is a passive declare, which is the only read-only question AMQP offers.

DeleteQueue takes the messages with it, so it is for tests and for tools. A service that deletes queues has usually confused a queue with a session. Deleting one that is not there is not an error: the caller asked for it to be gone and it is, which is what makes cleaning up after a test something other than guessing what ran.

Where to declare

At start-up, in the service that owns the queue, before consuming from it. Declaring on every publish costs a round trip; declaring nowhere means the first deployment to a fresh broker fails. Consumers declare their own queues and bindings; publishers declare the exchange they publish to.