Publishing

A publisher per destination

orders = mq.publisher("orders-events", "order.placed")
await orders.send({"orderId": "o-1", "totalCents": 4250})

Connection.publisher takes an exchange and a routing key, and both default to the empty string. Publishing to the default exchange — no exchange, routing key equal to a queue name — is how a message goes straight to one queue:

direct = mq.publisher(routing_key="shipping.orders")

Build them at start-up and keep them. A Publisher holds a destination, a codec and two flags; it opens nothing, registers nothing and is safe to share across tasks, so one per message type costs nothing and puts the decisions in one place. Building one per message is not wrong, only pointless.

It is not a coroutine

mq.publisher(...) is an ordinary method call and only send is awaited. That is deliberate: nothing about constructing a publisher touches the broker, and a coroutine there would suggest it did.

Sending

result = await orders.send({"orderId": "o-1", "totalCents": 4250})
result.message_id   # the envelope's id, which is also the AMQP message id
result.confirmed    # the broker acknowledged it, and is always true here:
                    # a refusal raises rather than coming back as a field
result.routed       # it reached at least one queue

The payload is whatever the codec can encode. With the default JsonCodec that is a dict, a list, a scalar — or a dataclass, which goes on the wire as an object with its field names:

@dataclass
class OrderPlaced:
    order_id: str
    total_cents: int


await orders.send(OrderPlaced("o-1", 4250))

A publisher bound to an exchange rather than to one key can name the key per message:

events = mq.publisher("orders-events")
await events.send(payload, routing_key="order.cancelled")

What the envelope gets

send builds an envelope when none is given: a fresh identifier, a correlation identifier defaulting to it, this connection's origin, attempt 1, version 1, and the time now. The type falls back to the routing key, which is why a publisher bound to order.placed needs no further configuration to say what it sends.

Carrying context forward

A message published because of another one should say so, or a trace through four services has four unrelated identifiers in it:

async def handle(message: Message) -> Ack:
    await shipped.send(
        {"orderId": message.payload["orderId"]},
        envelope=message.envelope.with_(
            causation_id=message.envelope.id,
            type="order.shipped",
        ),
    )
    return accept()

with_ returns a copy, because Envelope is frozen: an envelope describes a message that has already been published or received, and changing one in place would change what a log line said after it was written.

Carrying the envelope forward keeps correlation_id — the field the whole chain shares — and causation_id names the immediate parent. Note what is not carried: id should be new for a new message, so pass id=str(uuid.uuid4()) or build a fresh Envelope and copy the correlation across:

Envelope(
    type="order.shipped",
    correlation_id=message.envelope.correlation_id,
    causation_id=message.envelope.id,
)

origin is filled in for you either way: an envelope arriving without one gets this connection's, because a reply is published by this process and an origin naming the service before it would be a lie in a log line somebody is going to trust.

Your own headers

Application headers live on the envelope, apart from the reserved ones:

await orders.send(
    payload,
    envelope=Envelope(type="order.placed", headers={"tenant": "acme"}),
)

A reserved name — anything in the x-acemq- set — is refused rather than dropped, with a ValueError naming the offenders. Silently discarding a header somebody set is worse than saying no. See the reserved namespace.

For a header every message needs, do not put it at every call site. That is what interceptors are for:

async def stamped(context: PublishContext, send: PublishNext):
    context.set_header("tenant", tenant.current())
    return await send(context)


mq.intercept_publish(stamped)

Several at once

results = await orders.send_all([first, second, third])

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. A loop calling send waits for the broker between one message and the next, so a thousand messages on a link with 5 ms of latency cost five seconds of waiting that nothing needed. 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 rest of the batch is still awaited and then PublishError is raised 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. The named failure is named rather than printed: a failure with nothing in its message — the CancelledError every send gets when the connection closes underneath the batch is the common one — is reported by its class, so the sentence never ends The first failure was: and stops. __cause__ carries the first underlying failure in payload order — not the first one the broker answered — for callers that need to tell a rejection from a timeout:

try:
    results = await orders.send_all(batch)
except PublishError as failure:
    log.error("%s", failure)            # the counts
    log.error("%s", failure.__cause__)  # what went wrong first

How wide the batch gets on the wire is capped by the connection rather than by this method — see back pressure below. A list of a million is a million tasks and a thousand messages in flight, which is the difference between backpressure and a memory leak; a million tasks is still a million tasks, so chunk what comes out of a database rather than handing a cursor's worth of rows to one call.

The blocking API has the same method: sync.connect(...) gives a publisher whose send_all runs the batch on the loop thread and returns when every confirm is in.

Java spells this sendAll and .NET SendAllAsync, with the same ordering, the same partial-batch counts and the same wording in the failure. Go and Ruby have no equivalent yet.

Back pressure

max_outstanding_publishes — a thousand by default — caps how many publishes may be waiting for the broker at once, across the whole connection:

mq = await connect(
    "amqp://broker",
    max_outstanding_publishes=1000,          # the default
    confirm_timeout=timedelta(seconds=10),   # the default
)

The permit is taken before the message is written, which is the whole point: a bound applied after the write has already let the message into memory. Without one, a caller publishing faster than the broker confirms accumulates unconfirmed messages until the process dies — and that looks like throughput right up to the moment it does not.

It is the connection's bound and not a publisher's, so it counts everything: send, send_all, the retry rungs, the dead letters and a replay all come through one place. mq.outstanding_publishes is how many are waiting right now, and a number pinned at the ceiling is a broker that has stopped answering.

When every permit is taken and none comes free within confirm_timeout, the publish raises rather than waiting on:

acemq: cannot publish message 0f9c… to exchange '' with key 'orders.new':
1000 publishes are already waiting for a confirm and none completed within
0:00:10. The broker is not keeping up; publish more slowly rather than
buffering more.

That sentence is Java's word for word, so one runbook covers both. A publisher parked for ever behind a broker that has stopped answering is indistinguishable from a quiet service, which is the failure the deadline exists to prevent.

Java spells the setting maxOutstandingPublishes and .NET MaxOutstandingPublishes, both with the same default of a thousand. Java bounds its wait with confirmTimeout; .NET waits without one.

A blocked broker: refused, not failed

When RabbitMQ is low on memory or disk it blocks the connection and stops reading the socket until the alarm clears. A publish made while mq.blocked is True is refused at once with PublishingPausedError, before a permit is taken or a byte is written:

from acemq_amqp import PublishError, PublishingPausedError

try:
    await publisher.send(order)
except PublishingPausedError:
    # Declined, not lost: nothing was sent, so a retry cannot duplicate it.
    await asyncio.sleep(1)       # back off, shed load, or buffer
except PublishError:
    # Failed: the message may or may not have reached the broker.
    raise

It subclasses PublishError, so an existing except PublishError still catches it; catch it first to tell back pressure from a loss. Every other PublishError — a nack, an unroutable mandatory message, the permit deadline above — means the message may have been lost, and none of them is a PublishingPausedError. The permit deadline stays a failure on purpose: a broker that stopped confirming may be dead rather than busy, and Java, .NET and Go call it a failure too.

RabbitMQ only tells a connection it is blocked once it publishes, so the first publish after an alarm is written and then held with no confirm. It waits at most confirm_timeout (ten seconds by default) for that confirm and then raises a PublishError — failed, since it was written and may have arrived — instead of hanging until the alarm clears. The publish after it sees the block and is refused.

blocked being None — the state could not be read — is not treated as blocked. Go returns PublishingPausedError at the same point, .NET throws ConnectionBlockedException, and Java's standing load counts the same event as refused rather than failed.

Durability

transient = mq.publisher("metrics", "tick", persistent=False)

Messages are persistent by default: the broker writes them to disk and they survive a restart. persistent=False is for the ones where the next message makes the last one irrelevant — a gauge, a heartbeat, a cache invalidation on a short timer. It is faster, and it means a broker restart loses whatever was in the queue.

Persistence is not durability on its own. A persistent message on a non-durable queue still goes when the broker does; see exchanges, queues and bindings.

Messages that reach no queue

required = mq.publisher("orders-events", "order.placed", mandatory=True)
await required.send(payload)   # raises PublishError if nothing is bound

An unroutable message is the quietest failure AMQP has: the publish succeeds, the message is discarded, the consumer waits, and nothing anywhere says why. mandatory=True makes the broker return it, and this library raises PublishError rather than leaving the fact in the result, because a caller who does not read the result would otherwise carry on believing the message went somewhere:

from acemq_amqp import PublishError

try:
    await required.send(payload)
except PublishError as failure:
    log.error("%s went nowhere: %s", failure.message_id, failure)

Without mandatory, result.routed still says what happened — it is only the raising that is opted into.

A different codec for one publisher

from acemq_amqp import BytesCodec

raw = mq.publisher("files", "uploaded", codec=BytesCodec())
await raw.send(b"\x89PNG\r\n\x1a\n...")

The publisher's codec decides the content type on the wire, and a consumer picks its codec by reading that. See codecs.

What can go wrong

Most of what goes wrong with a broker is the broker's own exception, and this library does not wrap every one of them in a class of its own: that would hide the detail somebody actually needs while adding a name they then have to learn. The exceptions it does define are AceMQError, PublishError, PublishingPausedError and SecurityError, plus FatalError, which is a statement about a message rather than about the library.

Publisher confirms

Publisher confirms are on. send does not return until the broker has acknowledged the message, and result.confirmed says so. That is a round trip per message, which is the cost of knowing; a service that publishes in bulk should pipeline — send_all — rather than turning confirms off, because there is no way to turn them off here.

A refusal raises. If the broker declines to keep a message — a nack, which is different from an unroutable one — PublishError is raised:

cannot publish message o-1 to exchange "orders-events" with key "order.placed":
the broker refused to confirm it. The message reached the broker and the broker
has not taken responsibility for it, so it must be treated as not sent

So result.confirmed is always true on a result you are holding; a false one never reaches you. That is deliberate, and it was not always so. Until 0.7.2 a nack came back as confirmed=False and raised nothing, which meant a caller who did not inspect the field carried on believing the message had gone — and the outbox relay, which removes a record when the publish does not raise, deleted records for messages the broker had refused. Every other library in this family raises on a nack; now so does this one.

For the stronger guarantee — that a message and the database row it describes either both happen or neither does — see the outbox.