Publishing

mq.publish({ "order_id" => "A-1" }, "order.placed",
           "orders-events", "order.placed.v2")

One method. The payload is a Ruby object, to: is the routing key, exchange: is where it goes, and everything after that describes the envelope.

Where a message goes

exchange: defaults to "", the default exchange, which routes by queue name: to: is then the name of a queue. That is convenient for a first test and a poor arrangement between services, because the publisher has to know the name of the consumer's queue. A named exchange with a binding lets the consumer decide what it wants, and lets a second consumer arrive without the publisher changing.

mq.publish(event, "orders.new")                                 # default exchange
mq.publish(event, "order.placed", "orders-events")    # topic exchange

The default exchange also cannot be bound, listed or given a policy, which is the same reason retry rungs come home through acemq.retry rather than through it.

The envelope

Every message carries one. Left alone, publish builds it: a fresh UUID for id, correlation_id equal to that id, attempt 1, first_seen now, and the connection's origin.

Keyword arguments after exchange: set its fields:

mq.publish(event, "order.placed", "orders-events",
           "order.placed.v2",
           2,
           incoming.envelope.correlation_id,
           incoming.envelope.id,
           { "tenant" => "acme" })

headers: is your own; the reserved x-acemq- names are refused rather than silently dropped. The envelope has the full list and what each field is for.

For a message whose metadata derives from another's, build the envelope and pass it:

outgoing = incoming.envelope.with("shipment.requested",
                                  incoming.envelope.id)
mq.publish(payload, "shipment.requested", "shipping-events",
           outgoing)

Envelope is frozen, so with returns a new one. Passing both an envelope and the fields to build one raises ArgumentError — one or the other, because the two answers differ and neither is obviously the one that was meant.

What publish returns

The envelope that actually went on the wire — what the interceptors left, not what you handed in. An interceptor that stamps a tenant or rewrites the routing key has already run by then, so this is the id to log if you are going to log one:

sent = mq.publish(event, "order.placed", "orders-events")
logger.info("published #{sent.id} correlation=#{sent.correlation_id}")

Durability

persistent: true is the default: the broker is asked to write the message to disk, so a broker restart does not lose it. Pass persistent: false for something whose value expires faster than a restart takes — a cache invalidation, a heartbeat — and remember that durability is a property of the message and the queue. A persistent message on a transient queue is still lost with the queue.

Codecs

The connection's codec turns the payload into bytes and sets the content type. JSON unless you said otherwise, and per-publish when one message is different:

mq = Connection.open(url, StringCodec.new)

mq.publish(png_bytes, "thumbnails", BytesCodec.new)

See codecs for the four that ship, how a composite one decodes several content types, and what it takes to write your own.

Failures

publish raises. A broker that cannot be reached, a payload no codec can encode, an interceptor that refused the message — all of them come out as an exception, and none of them is quietly swallowed, because a publisher that returns normally having sent nothing is the failure mode you find out about from the consumer's absence.

begin
  mq.publish(event, "order.placed", "orders-events")
rescue AceMQ::AMQP::PublishError => e
  # the broker did not confirm it
rescue AceMQ::AMQP::EncodeError => e
  # the codec could not turn this payload into bytes
end

Publishing is confirmed: the transport waits for the broker to acknowledge the message and raises PublishError if it does not. A publish that returns has been accepted by the broker — not necessarily routed to a queue, which is a separate thing an unbound exchange will do silently, and which the next section is about.

Whatever the failure, acemq.publish.total{outcome="failed"} is counted before the interceptors are told — except a publish the library declined before writing anything, below, which is outcome="refused" because it cannot have been lost — so a publish an interceptor refused is counted too. It did not reach the broker, which is what the metric is about. See metrics and health.

While the connection is being recovered

One failure is worth knowing by name, because it is brief, expected, and the right response to it is a retry rather than an alert:

the connection is being recovered and its channels are not open yet. Publishing
now would be refused by the broker and would close the connection again; this
message was not sent, and a retry once the recovery completes will go down the
recovered connection.

When a broker node restarts or the network drops, bunny rebuilds the connection underneath you. That happens in two steps — the socket and the AMQP handshake first, the channels after — and a message written in between would arrive on a channel the broker has not opened yet. The broker's answer to that is not to refuse the message; it closes the whole connection, which would undo the recovery in progress and start another one.

So publishing is refused for the length of the window and the message is not sent. The window opens when bunny notices the loss, not when its recovery attempt starts: bunny first sleeps network_recovery_interval (5s by default), and a publish in those seconds is refused as well, with

the connection has been lost and is waiting to be recovered; this message was
not sent, and a retry once the recovery completes will go down the recovered
connection.

With a single broker address the whole window is one network_recovery_interval and the reconnect: a forced close on the drill cluster, on bunny 2.24 and 3.4 alike, stopped publishing about 0.65s after the close (when bunny noticed) and had it going again about 5.05s later. Lowering network_recovery_interval (a keyword on Connection.open) shortens it one for one. Retry with a back-off: by the time a retry lands the channels are back. See recovery for why it was twice that up to 0.7.7. Both refusals are a PublishingPausedError -- a PublishError, so code that already retries needs no change -- and both count as refused, never failed. So does a publish bunny itself would not write because the connection closed between the check and the write (Bunny::ConnectionClosedError or Bunny::ChannelAlreadyClosed from basic_publish): nothing reached the socket. Up to 0.7.7 the seconds before the attempt went to bunny that way and came back as a plain PublishError, counted failed -- about 85 per recovery on the drill cluster, every one unsent.

A publish that was waiting for its confirm when the connection dropped is a different case: the message was written and nothing answered for it, so it may or may not have arrived, and it fails as an ordinary PublishError — not as a refusal. bunny itself never ends that wait: its recovery replaces the queue the waiting thread is parked on, so the wait runs to bunny's whole continuation_timeout (15s), and on bunny 2.24 and 3.4 alike every other publisher on the connection stalled behind it, five seconds past the recovery. The transport ends the wait as soon as bunny starts its recovery attempt, one network_recovery_interval after the loss, which brought that stall from 15.06s to 5.06s. A lower network_recovery_interval shortens it further.

Consuming is unaffected — deliveries resume on their own once the channels are recovered.

While the broker has blocked the connection

RabbitMQ blocks a publishing connection when it is low on memory or disk, and from then on it stops reading the socket. A publish written into that socket is not refused — it waits, and if bunny gives up waiting it comes back as an unconfirmed message, which is the same PublishError a message the broker may have lost raises. A caller cannot tell back pressure from loss that way, and should not have to.

So a publish on a connection that is already blocked is declined before anything is written, with its own error:

begin
  mq.publish(event, "order.placed", "orders-events")
rescue AceMQ::AMQP::PublishingPausedError => e
  # not sent: the broker is low on memory or disk. Safe to send again later.
  back_off_and_retry(event)
rescue AceMQ::AMQP::PublishError => e
  # may have been lost: the broker did not confirm it
end

PublishingPausedError is a subclass of PublishError, so code that rescues PublishError keeps working and simply treats back pressure as before. It is counted as acemq.publish.total{outcome="refused"}, and the publish span says refused, so a broker alarm does not show up in the failed series that means a message may have been lost. A publish declined during a recovery is the same error and the same word. The message carries the broker's reason (low on memory, low on disk space). It is the counterpart of Go's isPaused, .NET's ConnectionBlockedException and Java's PublishingPausedException.

publish_all on a blocked connection answers every message the same way, and the batch raises PublishingPausedError only when every failure in it was declined unsent — one message that may have been lost makes it a plain PublishError, because blindly resending that batch could duplicate.

A block that arrives after a publish has been written cannot be undone: that publish waits as it always has, and is confirmed once the broker reads the socket again. mq.blocked? and mq.blocked_reason say whether the connection is blocked now, without a round trip.

When reaching no queue should be an error

A confirm says the broker has the message. It does not say the message reached a queue, and the two come apart more often than they look like they should: an exchange with no matching binding, a typo in a routing key, a consumer's queue that was never declared. The publish is confirmed and the message is dropped in the same breath. It is the quietest failure AMQP has — the publisher succeeded, the consumer is still waiting, and nothing anywhere says why.

mandatory: true asks the broker to hand the message back instead:

begin
  mq.publish(event, "order.placed", "orders-events", true)
rescue AceMQ::AMQP::PublishError => e
  raise unless e.unroutable?

  logger.error("nothing is bound to order.placed on orders-events: #{e.message}")
end

PublishError#unroutable? is what separates the two: false is a message the broker would not take, true is one it took and could not route. One exception class with a flag rather than two classes, so a caller who only wants to know that the message did not arrive still rescues one thing. Go and Python split the same failure the same way; Java has it as PublishOptions.allowUnroutable(), which is the same choice made the other way up.

It is off by default, and per publish. Turning it on for everybody would turn a message nobody happens to be listening for yet into an exception in code that has never seen one, which is a fair description of most first deployments. It also costs a round trip only when a message really is unroutable: the return frame arrives ahead of the confirm the publish was already waiting for, so a message that routes normally pays nothing.

An unroutable publish counts acemq.publish.total{outcome="unroutable"} rather than outcome="failed", and marks the publish span unroutable, which is an error outcome — with the broker's own reply text as messaging.acemq.reason. The word is kept apart from failed because they are fixed in different places: failed is a broker or a network, unroutable is a binding nobody made.

The library uses this on its own account, everywhere it publishes for you and then treats the work as done: a retry, a message set aside into {queue}.dlq or {queue}.parked, the outbox relay, a routing slip's or then_publish's hop to the next step, the scheduler's hops and final delivery, and replay. Each of those keeps the original — retried, handed back or left in place — when the broker had nowhere to put the copy, rather than acknowledging it into nothing. See when a message cannot be set aside. Your own publish keeps its default.

The one exception is a responder's reply. It is not mandatory: a reply queue that has gone means the caller has gone, and retrying the request would only repeat the work for nobody. A reply lost that way shows up on the requester's side as a timeout.

Publishing a batch

publish waits for the broker to confirm each message before it returns, which is the right trade for one message and the wrong one for a thousand: a thousand messages cost a thousand round trips, and the publisher spends nearly all of that time waiting for a network. publish_all hands every message over first and waits for all of the confirms together:

envelopes = mq.publish_all(orders, "order.placed", "orders-events",
                           "order.placed.v2")

The envelopes come back in the order the payloads were given, whatever order the broker confirmed them in. Everything else works as it does for a single publish: each message gets an envelope of its own (or pass envelopes:, one per payload), each one goes through the publish interceptors exactly once, each one is counted in acemq.publish.total on its own, and mandatory: true means for each message what it means for one.

It is not atomic, and nothing in AMQP could make it so. There is no way to publish a hundred messages such that all or none arrive, and a library that offered one would be lying. What this does instead is tell you how much did arrive:

begin
  mq.publish_all(orders, "order.placed", "orders-events")
rescue AceMQ::AMQP::PublishError => e
  # "3 of 500 messages were not confirmed; 497 were. The first failure was: ..."
  logger.error(e.message)
end

A batch that half succeeded is the ordinary outcome of a broker problem partway through, so every message is still waited for after the first failure and the counts in that sentence are real. A caller told only "it failed" republishes hundreds of messages that are already on a queue. The failure quoted is the first one in payload order rather than the first the broker answered, so the same half-failed batch reports the same message twice running. Java's sendAll and .NET's SendAllAsync raise that sentence word for word, so one runbook covers all three.

unroutable? on that error is true only when every failure was a message the broker had nowhere to route — that being the only case where concluding "nothing is bound" is right. A batch that met a missing binding and a refused message is not an unroutable batch.

How many messages may be unconfirmed at once

A thousand, by default, for the whole connection:

mq = AceMQ::AMQP::Connection.open("amqp://localhost",
                                  1000) # the default

That ceiling is the difference between backpressure and a memory leak that looks like throughput. A publisher that hands over messages faster than the broker confirms them is accumulating unconfirmed messages in this process and in the broker, and nothing about the accumulation is visible until the process dies of it. Java's maxOutstandingPublishes and .NET's MaxOutstandingPublishes are the same number in the same place.

publish_all respects it by writing in waves: up to the ceiling goes out, those confirms come back, and only then is the next wave written. A batch that fits inside the ceiling — which is nearly all of them — is one wave and is exactly as fast as it always was. A batch of a hundred thousand is a hundred waves rather than a hundred thousand messages nobody bounded, and you no longer have to split it by hand. The results still come back in payload order and each message still gets its own answer.

Room is given back when the broker answers for a message, and an ack, a nack and a basic.return are all answers. A message the broker has taken and said nothing about keeps its room, because that message really is still outstanding. So when a broker stops confirming altogether, the ceiling fills and the messages that cannot be written are told why rather than buffered:

cannot publish message m-4 to exchange "orders-events" with key "order.placed":
1000 publishes are already waiting for a confirm and none of them completed.
The broker is not keeping up; publish more slowly rather than buffering more.

Raise the ceiling if you have measured that you need to and have the memory for it. Reaching it is a signal about the broker, and buffering more is the one response to that signal which cannot help.

Publishing and a database in the same breath

A service that writes a row and then publishes has two things that can fail independently, and there is no arrangement of begin/ensure that closes the gap. That is what the outbox is for: the message is written in the same transaction as the work, and a relay publishes what was committed.

Threading

publish is safe to call from several threads. The transport keeps one publishing channel, opened once with confirms enabled, and holds a mutex across the publish and its confirm — a bunny channel is not safe to use from two threads at once, and a consumer thread dead-lettering a message publishes on that same channel while an application thread may be publishing its own. The cost is that publishes on one connection serialise; the alternative is two threads interleaving frames into a protocol error.

publish_all holds that same mutex for the whole batch — every wave and every wait — because the delivery tags it hands back have to belong to this batch and to nothing else. A very large batch therefore keeps other threads on this connection waiting for as long as it takes. The ceiling bounds the memory a batch costs, not the time it holds the mutex for; if other threads on this connection must keep publishing while a long batch runs, give the batch a connection of its own.

Interceptors are called on whichever thread is publishing, so one that keeps state has to be safe to call from several at once.

Next