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

One AMQP connection + confirm-mode channel per `{instance, url, exchange}`.
Declares the configured exchange whenever it opens the channel (idempotent) —
never a queue; queue ownership belongs to consumers, not this sink.

Every publish is `mandatory` and is answered on the broker's confirm of that
publish, so `publish/4` answers `:ok` only when at least one queue bound to
the exchange accepted the message. A message the exchange routes to no queue
is `{:error, {:unroutable, routing_key}}`, one the broker refuses (a queue's
`reject-publish` overflow) is `{:error, :nacked}`, and no confirm within
`:confirm_timeout_ms` milliseconds is `{:error, :confirm_timeout}`.

## Confirms are asynchronous

A publish is sent and its caller parked under the publish's sequence number;
the confirm, whenever it arrives, answers it. Many publishes are in flight on
the channel at once (up to `:max_inflight`, default 256; past that a publish
is `{:error, :busy}`), so one slow confirm never serializes the others.

Every call carries its caller's deadline. A publish still waiting in this
process's mailbox when its caller gave up (`publish/4` waits 15 s) is
answered `{:error, :expired}` and never sent: an abandoned call is not
published behind the caller's back.

The channel is monitored. When the broker closes it (a publish to a deleted
exchange is a 404) every publish waiting on it answers
`{:error, {:channel_closed, reason}}` and the channel is reopened at once on
the same connection, re-declaring the exchange.

The connection dials after `start_link/1` returns, and connection loss is
not treated as a crash — this GenServer stays up and retries on a timer,
replying `{:error, :not_connected}` to publishes in the meantime (and to any
that were waiting when it dropped). Every one of these errors flows into the
source's `Ankusa.RetryPolicy` exactly like any other sink failure, so there
is no separate reconnect policy to get wrong.

The AMQP connection belongs to `amqp_client`'s own supervisor, not to this
process, so stopping this process (its supervisor's shutdown, or the idle
sweep of `Ankusa.Sink.Reaper`) closes it explicitly and answers every
waiting publish `{:error, :not_connected}`.

# `call_timeout_ms`

```elixir
@spec call_timeout_ms() :: pos_integer()
```

The longest `publish/4` waits, in milliseconds.

# `child_spec`

Returns a specification to start this module under a supervisor.

See `Supervisor`.

# `default_url`

```elixir
@spec default_url() :: String.t()
```

The AMQP URL this connection uses when `:url` is not configured.

# `publish`

```elixir
@spec publish(GenServer.server(), String.t(), binary(), keyword()) ::
  :ok | {:error, term()}
```

# `start_link`

---

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