# `Ankusa.Sink.RabbitMQ`
[🔗](https://github.com/jamescarr/ankusa/blob/ankusa_rabbitmq-v0.7.0/packages/ankusa_rabbitmq/lib/ankusa/sink/rabbitmq.ex#L1)

Publishes delivered hooks to a RabbitMQ **exchange**. Never touches a queue
— binding a queue to the exchange, and everything downstream of that, is
the consumer's job, not this sink's. That mirrors real AMQP topology
ownership: producers own exchanges, consumers own their own queues.

Messages are small on purpose: the body rides inline up to
`:inline_max_bytes` (default 64 KiB) and is checked in through
`Ankusa.ClaimCheck` above that, so RabbitMQ throughput and memory stay flat
regardless of payload size, and any consumer — BEAM or not — redeems the
ticket without blob-store credentials of its own. The message is
`Ankusa.Sink.Message`, byte-identical to what `Ankusa.Sink.Kafka` produces.

`deliver/3` answers `:ok` only when the broker confirmed the message and at
least one queue bound to the exchange accepted it: every publish is
`mandatory`, so a message the exchange routes to no queue is
`{:error, {:unroutable, routing_key}}`, retried by the source's
`Ankusa.RetryPolicy` and then dead-lettered like any other sink failure.
Every publish carries the hook's `id` as AMQP `message_id`, plus the
`ankusa_idempotency_key` header (the key consumers dedupe on, the message
JSON's `idempotency_key`) and the `ankusa_dedupe_key` and `ankusa_replay_id`
headers when the hook carries them, so consumers dedupe on the same identity
the message JSON exposes.
Surviving a broker restart is the queue's property: messages are always
published `persistent`, and durable classic and quorum queues persist them
before confirming. The other errors are `{:error, :nacked}` (the broker
refused it, e.g. a queue's `reject-publish` overflow),
`{:error, :confirm_timeout}`, `{:error, {:channel_closed, reason}}` (the
channel is reopened at once, re-declaring the exchange),
`{:error, {:publish_failed, reason}}` and `{:error, :not_connected}`.

## opts

  * `:exchange`          — required
  * `:exchange_type`     — default `:topic`
  * `:url`                — AMQP URL, default `"amqp://guest:guest@localhost:5672"`
  * `:routing_key`       — a static string, or a 1-arity fun `(Envelope.t() -> String.t())`;
                            default `"ankusa.#{source_id}"`
  * `:inline_max_bytes`  — default 64 KiB (65,536), configurable
  * `:max_message_bytes` — the largest message published; default
                           `134_217_728`, RabbitMQ's default `max_message_size`
                           (128 MiB). Lower it to the broker's own setting
  * `:retry_ms`          — reconnect backoff in milliseconds, default `5_000`
  * `:confirm_timeout_ms` — publisher-confirm wait in milliseconds, default `5_000`
  * `:idle_timeout_ms`   — close the connection after this long without a
                            delivery, default `600_000` (10 min); `0` never.
                            Never under the publish call's 15 s plus 1 s.
                            See `Ankusa.Sink.Reaper`

---

*Consult [api-reference.md](api-reference.md) for complete listing*