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, default5_000:confirm_timeout_ms— publisher-confirm wait in milliseconds, default5_000:idle_timeout_ms— close the connection after this long without adelivery, default `600_000` (10 min); `0` never. Never under the publish call's 15 s plus 1 s. See `Ankusa.Sink.Reaper`