Publishing

A publisher sends payloads of one type to one exchange and routing key. Create it once and keep it: it holds no connection of its own, but it does hold the back pressure that stops a loop from queueing more unconfirmed publishes than the broker will ever confirm.

using var mq = await AceMqConnection.ConnectAsync("amqp://localhost");
var publisher = mq.Publisher<OrderPlaced>("orders", "order.placed");

PublishResult result = await publisher.SendAsync(new OrderPlaced("A-1", 42.50m));

PublishResult carries the envelope id the message went out with, whether the broker routed it, and how long the confirmation took.

What "sent" means

SendAsync returns after the broker has confirmed the message, not after the bytes reach the socket. Those are different events, and the gap between them is where messages are lost on a broker restart.

Confirms are on by default. Turning them off is a deliberate act:

var config = ConnectionConfig.ForUrl("amqp://localhost")
    .WithoutPublisherConfirms()
    .Build();

Without them a publish reports success as soon as it is written. That is faster and it is a weaker promise; the library will not make it quietly on your behalf.

That applies to your publishes. What the library publishes for you is confirmed either way: a retry hop, a move to {queue}.dlq or {queue}.parked, a replay, a routing-slip or pipeline hop, a scheduled hop or delivery, an outbox record, a responder's reply. Each of those acknowledges a message or marks a record done once the publish returns, and without a confirm a broker that dropped the message as unroutable is indistinguishable from one that stored it. So a connection made without confirms opens one more channel, in confirm mode and watched for returns, and sends only those on it. Your own publishers stay unconfirmed and report every send as routed, as before.

The cost is the confirm round trip on each of those hops — the same as with confirms on. Measured against a local RabbitMQ 4, 500 messages retried five times each: about 16,000 hops a second unconfirmed, about 1,100 confirmed.

Unroutable messages fail

A message published to an exchange with no matching binding is a failure, not a silent discard:

PublishFailedException: 7d3f… reached the broker but matched no queue bound to
'orders' for 'order.placed'. This is usually a topology mistake; if the discard is
intended, publish with PublishOptions.AllowUnroutable().

This is almost always a binding that was never declared, or a routing key with a typo in it. The cheapest moment to discover that is the publish call. The alternative — the broker's default — is that the message disappears and the problem surfaces hours later as an absence nobody can account for.

Where the discard really is intended:

var publisher = mq.Publisher<Telemetry>(
    "metrics", "cpu.sample", PublishOptions.Defaults().AllowUnroutable());

Options

PublishOptions.Defaults()            // persistent, unroutable reported
PublishOptions.TransientDelivery()   // not written to disk
    .ExpiringAfter(TimeSpan.FromMinutes(5))
    .WithPriority(4)

TransientDelivery is reasonable for a reading that is superseded within seconds. It is not reasonable for anything a person would notice the absence of, because a broker restart takes those messages with it.

Sending an envelope you built

The default envelope is created for you, with the type defaulted to the routing key and the correlation id defaulted to the message id. Supply your own to set correlation, causation or application headers:

var envelope = Envelope.Of("order.placed")
    .CorrelationId(incoming.Envelope.CorrelationId)   // continue the trace
    .CausationId(incoming.Envelope.Id)                // this message caused that one
    .Header("x-tenant", "acme")
    .Build();

await publisher.SendAsync(order, envelope);

Carrying the correlation id from the message you are handling is what makes a request traceable across services. Setting causation to the incoming message's id is what makes the chain reconstructable afterwards.

Application headers must not use the x-acemq- prefix — that namespace is the engine's, and writing into it throws rather than letting the header vanish on the way back out. See the envelope.

Several at once

IReadOnlyList<PublishResult> results = await publisher.SendAllAsync(orders);

Every message goes out before any confirm is awaited, and only then are all of them checked together — the throughput of pipelining with the safety of having waited. The results come back in the order the payloads did, whatever order the broker answers in.

This is not atomic. AMQP has no such thing: there is no way to publish a hundred messages such that all or none arrive, and a library that offered one would be lying. If any message fails, the whole batch is still awaited and then PublishFailedException is thrown naming how many were not confirmed and how many were:

2 of 100 messages were not confirmed; 98 were. The first failure was: ...

The counts are the point. A batch that half succeeded is the ordinary outcome of a broker problem partway through, and a caller told only "it failed" resends messages that already arrived. InnerException carries the first underlying failure, in payload order, for callers that need to tell a rejection from a timeout.

The pipelining is real all the way down. The RabbitMQ transport holds its publish lock across the frame write and releases it before waiting for the confirm, so a batch of a hundred is a hundred writes and then one wait rather than a hundred round trips. Two hundred messages against a loopback broker take about 6 ms as a batch and about 250 ms one at a time; on a link with real latency the gap is the latency multiplied by the batch size.

Back pressure

MaxOutstandingPublishes — 1000 by default — caps how many publishes may be waiting for a confirm at once. Reached, the next SendAsync waits for an answer before it writes. Without that cap, a caller in a tight loop can outrun the broker's ability to confirm, and the failure arrives as memory growth rather than as a slow publish.

It is enforced twice, and deliberately: once above the codec, where it has always been, and once inside the RabbitMQ transport immediately before the frame write. The second one is what makes the number mean something for anything driving the transport directly, and it is the bound the pipelining above is safe because of — the setting used to be irrelevant under a lock that allowed exactly one unconfirmed publish anyway.

When the broker stops accepting

A broker under a memory or disk alarm blocks its publishers. That state is visible rather than hidden as a hang:

if (mq.IsBlocked) Console.WriteLine(mq.BlockedReason);

A publish attempted on a blocked connection throws ConnectionBlockedException, and is counted as outcome="refused": nothing was written, so the message is still yours to send again.

mq.Health() reports a blocked connection as Up, with the reason on it — back pressure is worth an alert and is not worth a restart. See a blocked connection is Up, with the reason.